Object observer, Iterator, Generator, Observable

Reactive Programming

Reactive programming is programming with asynchronous data streams such as AJAX and Click events.

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

              single items               multiple items
synchronous T getData()               Iterable<T> getData()
asynchronous Future<T> getData() Observable<T> getData()

An Observable is the asynchronous/push “dual” to the synchronous/pull Iterable
event         Iterable (pull)         Observable (push)
retrieve data T next()                 onNext(T)
discover error throws Exception onError(Exception)
complete         !hasNext()         onCompleted()

"Rx = Observables + LINQ + Schedulers"

Rx.Observable.prototype.flatMapLatest(selector, [thisArg]);

var requestStream = Rx.Observable.just('https://api.github.com/users');
requestStream.subscribe(function(requestUrl) {
  // execute the request
  var responseStream = Rx.Observable.create(function (observer) {
    jQuery.getJSON(requestUrl)
    .done(function(response) { observer.onNext(response); })
    .fail(function(jqXHR, status, error) { observer.onError(error); })
    .always(function() { observer.onCompleted(); });
  });
  responseStream.subscribe(function(response) {
    // do something with the response
  });
}

------------------------------------------------

// Object observer
var person = {name: 'Bob', sex: 'Male'};

Object.observe(person, chagesHandler);
Object.unobserve(person, chagesHandler);

------------------------------------------------

// Iterator
function* iterateNumbers () {
    yeild 42;
    yeild 32;
    return 19;
}

function iterateNumbers () {
    var state = 0;
    return {
        next: function () {
            switch (state) {
                case 0:
                    state = 1;
                    return {value: 42, done: false};
                case 1:
                    state = 2;
                    return {value: 32, done: false};
                default:
                    return {value: 19, done: true};
            }
        }
    };
    yeild 42;
    yeild 32;
    return 19;
}

var iterator = iterateNumbers();
iterator.next(); // {value: 42, done: false}
iterator.next(); // {value: 32, done: false}
iterator.next(); // {value: 19, done: true}

while (iterator.next().done !== false) {
    console.log(iterator.next().value);
}

------------------------------------------------

// Generator - Async iteration

function getResultFormAsyncAJAX(url) {
    var promise = new Promise();
    $.ajax({
          url: url
        , type: 'GET'
        , data: null
        , dataType: 'json'
        , success: function (data){promise.resolve(data);}
        , error: function () {promise.error();}
    });
    return promise;
}

var generator = getResultFormAsyncAJAX('/some/path/to/file.html');
var result = generator.next(); // promise.next();
     result.then(function (data) {...}, function (error) {...});
     result = generator.next(data); // Генератор накапливает в себе полученные ответы на предыдущих шагах перебора
                               // тут next(data) вызывает следующий yeild итератора
     result.then(function (data) {...}, function (error) {...});
     result = generator.next(data);

task.js = generators + promises
regenerator - async functions

------------------------------------------------

// Observable
producer.observer(observer);
observer.next(42); // 42
observer.next(32); // 32
observer.return(); // done

nums().observer({
    next (v) {console.log(v);},
    return (v) {console.log('done' + v);},
    throw (e) {console.error(e);}
});

nums()
    .forEach(function (v) {console.log(v);}) // next
    .then(function (v) {console.log('done:' = v);}); // return

------------------------------------------------

Implementing Observable

WebSocket.prototype.observer = function (generator) {

    var cleanUp
        , message = function (v) {
            var pair = generator.next()'
            if (pair.done) {cleanUp();}
        }
        , error = function (e) {
            cleanUp();
            generator.throw(e);
        }
        , close = function (v) {
            cleanUp();
            generator.return(v);
        }

    function cleanUp () {
        this.removeEventListener('message', message);
        this.removeEventListener('error', message);
        this.removeEventListener('error', message);
    }

    this.addEventListener('message', message);
    this.addEventListener('error', message);
    this.addEventListener('error', message);

};

(async function(){
    for (var member on new WebSocket('/signups')) {
        console.log(JSON.stringify(num));
    }
})();

async function drag (element) {
    var downs = Observable.fromEvent(element, 'mousedown');
    var ups = Observable.fromEvent(document, 'mouseup');
    var moves = Observable.fromEvent(document, 'mousemoves');
    var drags = downs
                               .map(function (down) {moves.takeUntil(ups)})
                               .flatten();
    for (drag on drags) {
        element.style.top = drag.offsetY;
        element.style.left = drage.offsetX;
    }
}

------------------------------------------------

|-----------|Synchronous |Asynchronous|

|function  |T                   | Promise         |
|function*|Iterator          |Observable     |