RxJS -- Subscription

Subscription是什麼?

當subscribe一個observable的時候, 返回的就是一個subscription. 它是一個一次性對象(disposable), 它有一個很是重要的方法 ubsubscribe(), 它沒有參數, 它會dispose掉subscription所持有的資源, 或者叫取消observable的執行.

第一個例子:

import { Observable } from "rxjs/Observable";
import { Subscription } from "rxjs/Subscription";
import 'rxjs/add/observable/interval';

const observable = Observable.interval(1000);

const subscription = observable.subscribe(x => console.log(x));

console.log(subscription);

subscription.unsubscribe();

console.log(subscription);

 

運行結果是這樣的:

Subscriber {
  closed: false,
  _parent: null,
  _parents: null,
  _subscriptions: [ AsyncAction { closed: false, _parent: [Circular], _parents: null, _subscriptions: null, scheduler: [AsyncScheduler], work: [Function], pending: true, state: [Object], delay: 1000, id: [Timeout] } ],
  syncErrorValue: null,
  syncErrorThrown: false,
  syncErrorThrowable: false,
  isStopped: false,
  destination: 
   SafeSubscriber {
     closed: false,
     _parent: null,
     _parents: null,
     _subscriptions: null,
     syncErrorValue: null,
     syncErrorThrown: false,
     syncErrorThrowable: false,
     isStopped: false,
     destination: 
      { closed: true,
        next: [Function: next],
        error: [Function: error],
        complete: [Function: complete] },
     _parentSubscriber: [Circular],
     _context: [Circular],
     _next: [Function],
     _error: undefined,
     _complete: undefined } }
Subscriber {
  closed: true,
  _parent: null,
  _parents: null,
  _subscriptions: null,
  syncErrorValue: null,
  syncErrorThrown: false,
  syncErrorThrowable: false,
  isStopped: true,
  destination: 
   SafeSubscriber {
     closed: false,
     _parent: null,
     _parents: null,
     _subscriptions: null,
     syncErrorValue: null,
     syncErrorThrown: false,
     syncErrorThrowable: false,
     isStopped: false,
     destination: 
      { closed: true,
        next: [Function: next],
        error: [Function: error],
        complete: [Function: complete] },
     _parentSubscriber: [Circular],
     _context: [Circular],
     _next: [Function],
     _error: undefined,
     _complete: undefined } }

注意兩次控制檯輸出的closed屬性的值是不一樣的, true表示已經unsubscribe()了.

在ubsubscribe以後, _subscriptions屬性也變成空了, 以前它是一個數組, 說明subscription能夠是多個subscriptions的組合.

毀滅函數

若是使用Observable.create方法的話, 它的參數函數能夠返回一個function. 而subscription在unsubscribe這個observable的時候, 會調用這個參數函數返回的function, 看例子:

import { Observable } from "rxjs/Observable";
import { Subscription } from "rxjs/Subscription";
import 'rxjs/add/observable/interval';

const observable = Observable.create(observer => {

    let index = 1;
    setInterval(() => {
        observer.next(index++);
    }, 200);

    return () => { // 在這能夠作清理工做 console.log('我在Observable.create返回的function裏面...'); };
});

const subscription = observable.subscribe(
    x => console.log(x),
    err => console.error(err),
    () => console.log(`complete..`)
);

setTimeout(() => {
    subscription.unsubscribe();
}, 1100);

 

運行結果:

這個例子很好的解釋了我寫的那一堆拗口的解釋..

 

retry, retryWhen的原理

直接舉例:

import { Observable } from "rxjs/Observable";
import { Subscription } from "rxjs/Subscription";
import 'rxjs/add/observable/interval';
import 'rxjs/add/operator/retry';

const observable = Observable.create(observer => {

    setInterval(() => {
        observer.next('doing...');
        observer.error('error!!!');
    }, 200);

    return () => {
        // 在這能夠作清理工做
        console.log('我在Observable.create返回的function裏面...');
    };
}).retry(4);

observable.subscribe(
    x => console.log(x),
    err => console.error(err),
    () => console.log(`complete..`)
);

 

能夠看到, 每次執行next以後都會有錯誤, 重試4次.

運行結果:

能夠看到, retry/retryWhen其實的原理便是先unsubscribe而後再從新subscribe而已, 因此每次retry都會運行我所稱的毀滅函數.

 

操做多個Subscriptions

多個subscriptions能夠一塊兒操做, 一個subscription能夠同時unsubscribe多個subscriptions, 使用add方法爲subscription添加另外一個subscription. 對應的還有一個remove方法.

直接舉官網的例子:

var observable1 = Observable.interval(400);
var observable2 = Observable.interval(300);

var subscription = observable1.subscribe(x => console.log('first: ' + x));
var childSubscription = observable2.subscribe(x => console.log('second: ' + x));

subscription.add(childSubscription);

setTimeout(() => {
  // Unsubscribes BOTH subscription and childSubscription
  subscription.unsubscribe();
}, 1000);

 

運行結果: