Core Operators

1.流有取消和处理机制，是对Javascrpit原生事件系统的改进

Streams provide their own mechanisms for cancellation and disposal, which is an improvement over JavaScript’s native event system.

 const progressBar$ = Rx.Observable.create(observer => {
   const OFFSET = 3000;
   const SPEED =  50;

   let val = 0;
   let timeoutId = 0;
   function progress() {
     if(++val <= 100) {
       observer.next(val);
       timeoutId = setTimeout(progress, SPEED);
     }
     else {
       observer.complete();
     }
   };
   timeoutId  = setTimeout(progress, OFFSET);

   return () => { //#A
      clearTimeout(timeoutId);
    };
});

//--------------------------------------------------//
//                Usage                             //
//--------------------------------------------------//
const subs = progressBar$.subscribe(console.log, null, () => console.log('Complete!'));

setTimeout(() => {
  subs.unsubscribe();
}, 6000);

2.Observable数据类型允许能使用流畅的函数链，允许使用和array类似的operators

The Observable data type enables fluent function chaining that allows the sequential application of operators, using a model similar to that of arrays.

 Rx.Observable.from([
   'The quick brown fox',
   'jumps over the lazy dog'
   ])
   .map(str => str.split(' '))
   .do(arr => console.log(arr.length))
   .subscribe(console.log);

//4
//The,quick,brown,fox
//5
//jumps,over,the,lazy,dog

除了map,还有filter,reduce,scan，这和array的很相似

3.不同于Javascript原生的Promise，observable具有处理和取消的内置功能

Unlike JavaScript’s native promises, observables have built-in capabilities for disposal and cancellation.

const promise = new Promise((resolve, reject) => {   setTimeout(() => {
       resolve(42);
    }, 10000);
});
promise.then(val => {
   console.log(`In then(): ${val}`);
});
const subscription$ = Rx.Observable.fromPromise(promise).subscribe(val => {
   console.log(`In subscribe(): ${val}`);
});
subscription$.unsubscribe();

// Console:
// In then(): 42

由于取消了订阅，Observable的Promise没有执行。

4.operator注入的函数，包括应用的业务逻辑，而且无副作用

Functions injected into the operators of an observable sequence contain the business logic of your application and should be side effect–free.

function exclude(predicate) {
  return Rx.Observable.create(subscriber => {
      let source = this;
      return source.subscribe(value => {
           try {
              if(!predicate(value)) {
                  subscriber.next(value);
              }
           }
           catch(err) {
              subscriber.error(err);
           }
         },
         err => subscriber.error(err),
         () => subscriber.complete());
   });
}

Rx.Observable.prototype.exclude = exclude;

//--------------------------------------------------//
//                Usage                             //
//--------------------------------------------------//
Rx.Observable.from([1, 2, 3, 4, 5])
  .exclude(x => x % 2 === 0)
  .subscribe(console.log);

5.observable是self-contained的，可以无限链试使用operators

Observables are self-contained with indefinitely chainable operators

6.operator彼此独立，只在前面的operator上的输出上工作

Operators act independent of each other and work only on the output of the operator that preceded them.

7.operator 的顺序和类型，决定了observable的行为和表现特征

The order and type of operators used determine the behavior and the performance characteristics of an observable.

let candidates = [
    {name: 'Brendan Eich', experience : 'JavaScript Guru'},
    {name: 'Emmet Brown', experience: 'Historian'},
    {name: 'George Lucas', experience: 'Sci-fi writer'},
    {name: 'Alberto Perez', experience: 'Zumba Instructor'},
    {name: 'Bjarne Stroustrup', experience: 'C++ Developer'}
];



//--------------------------------------------------//
//                Usage                             //
//--------------------------------------------------//
Rx.Observable.from(candidates)
  .pluck('experience')
  .take(2)
  .do(val => console.log(`Visiting ${val}`))
  .subscribe();
  // prints "Visiting JavaScript Guru"
  //        "Visiting Historian"