ReactiveX編程範式

ReactiveX

http://reactivex.io/

An API for asynchronous programming
with observable streams

The Observer pattern done right

ReactiveX is a combination of the best ideas from
the Observer pattern, the Iterator pattern, and functional programming

Easily create event streams or data streams.

Compose and transform streams with query-like operators.

Subscribe to any observable stream to perform side effects.

 

 

http://reactivex.io/intro.html

ReactiveX is a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data and/or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.

 

The ReactiveX Observable model allows you to treat streams of asynchronous events with the same sort of simple, composable operations that you use for collections of data items like arrays. It frees you from tangled webs of callbacks, and thereby makes your code more readable and less prone to bugs.

 

http://reactivex.io/documentation/observable.html

In ReactiveX an observer subscribes to an Observable. Then that observer reacts to whatever item or sequence of items the Observable emits. This pattern facilitates concurrent operations because it does not need to block while waiting for the Observable to emit objects, but instead it creates a sentry in the form of an observer that stands ready to react appropriately at whatever future time the Observable does so.

 

ReactiveX中文

https://segmentfault.com/a/1190000003632186

響應式編程是一種面向數據流和變化傳播的編程範式，數據更新是相關聯的。好比不少時候，在寫界面的時候，咱們須要對事件作處理，伴隨着前端事件的增多，對於事件的處理愈發須要更加方便的處理。

設想一下，平時在處理事件的時候，一單上了複雜度，好比輸入的時候，須要中止輸入的時候才進行，這個時候又只能輸入長度大於2才進行事件，當仍是以前的數據的話不進行事件，能夠考慮一下這個狀況下如何去寫。

 

例子

var keyup = Rx.Observable.fromEvent($input, 'keyup')
      .map(function (e) {
        return e.target.value; 
      })
      .filter(function (text) {
        return text.length > 2; 
      })
      .debounce(750)
      .distinctUntilChanged(); 

三秒後解除

var btn = document.getElementById('button');
var logRun = Rx.Observable.fromEvent(btn, 'click')
             .merge(Rx.Observable.timer(3000))
             .subscribe(e => {
               console.log('run!');
               logRun.dispose(); // 若是是一次性的就移除observable
             });

 

 

ReactJS語言實現

https://github.com/Reactive-Extensions/RxJS

例子如上。

 

教程！！

http://xgrommx.github.io/rx-book/why_rx.html

爲何要使用 ReactJS， 由於 其將 Promise 和 DOM 做爲一個總體對待。

One question you may ask yourself, is why RxJS? What about Promises? Promises are good for solving asynchronous operations such as querying a service with an XMLHttpRequest, where the expected behavior is one value and then completion. The Reactive Extensions for JavaScript unifies both the world of Promises, callbacks as well as evented data such as DOM Input, Web Workers, Web Sockets. Once we have unified these concepts, this enables rich composition.

 

 

ReactLua語言實現

https://github.com/bjornbytes/RxLua

 

Reactive Extensions for Lua.

RxLua gives Lua the power of Observables, which are data structures that represent a stream of values that arrive over time. They're very handy when dealing with events, streams of data, asynchronous requests, and concurrency.

 

local Rx = require 'rx'

Rx.Observable.fromRange(1, 8)
  :filter(function(x) return x % 2 == 0 end)
  :concat(Rx.Observable.of('who do we appreciate'))
  :map(function(value) return value .. '!' end)
  :subscribe(print)

-- => 2! 4! 6! 8! who do we appreciate!

 

協程異步。

local Rx = require 'rx'
local scheduler = Rx.CooperativeScheduler.create()

-- Cheer someone on using functional reactive programming

local observable = Rx.Observable.fromCoroutine(function()
  for i = 2, 8, 2 do
    coroutine.yield(i)
  end

  return 'who do we appreciate'
end, scheduler)

observable
  :map(function(value) return value .. '!' end)
  :subscribe(print)

repeat
  scheduler:update()
until scheduler:isEmpty()

 

最簡潔

local Rx = require 'rx'

-- Create an observable that produces a single value and print it.
Rx.Observable.of(42):subscribe(print)

 

concat例子

local Rx = require 'rx'

local first = Rx.Observable.fromRange(3)
local second = Rx.Observable.fromRange(4, 6)
local third = Rx.Observable.fromRange(7, 11, 2)

first:concat(second, third):dump('concat')

print('Equivalent to:')


Rx.Observable.concat(first, second, third):dump('concat')

 

觀察者模式

local Rx = require 'rx'

local subject = Rx.Subject.create()

subject:subscribe(function(x)
  print('observer a ' .. x)
end)

subject:subscribe(function(x)
  print('observer b ' .. x)
end)

subject:onNext(1)
subject(2)
subject:onNext(3)

 

多道併發

local Rx = require 'rx'
local scheduler = Rx.CooperativeScheduler.create()
local timerResolution = .25
local function log(message)
  print('[' .. string.format('%.2f', scheduler.currentTime) .. '] ' .. message)
end

-- Demonstrate Rx.Scheduler.Cooperative by running some simultaneous cooperative threads.
scheduler:schedule(function()
  log('this is like a setTimeout')
end, 2)

scheduler:schedule(function()
  local i = 1
  while true do
    log('this prints i twice per second: ' .. i)
    i = i + 1
    coroutine.yield(.5)
  end
end)

scheduler:schedule(function()
  for i = 1, 3 do
    log('this will print for 3 updates after 1 second')
    coroutine.yield()
  end
end, 1)

-- Simulate 3 virtual seconds.
repeat
  scheduler:update(timerResolution)
until scheduler.currentTime >= 3