Redux 数据更新

  1. function observable() {
  2.   const outerSubscribe = subscribe
  3.   return {
  4.     /**
  5.        * The minimal observable subscription method.
  6.        * @param observer Any object that can be used as an observer.
  7.        * The observer object should have a `next` method.
  8.        * @returns An object with an `unsubscribe` method that can
  9.        * be used to unsubscribe the observable from the store, and prevent further
  10.        * emission of values from the observable.
  11.        */
  12.     subscribe(observer: unknown) {
  13.       if (typeof observer !== 'object' || observer === null) {
  14.         throw new TypeError('Expected the observer to be an object.')
  15.       }
  17.       function observeState() {
  18.         const observerAsObserver = observer as Observer<S>
  19.           if (observerAsObserver.next) {
  20.             observerAsObserver.next(getState())
  21.           }
  22.       }
  24.       observeState()
  25.       const unsubscribe = outerSubscribe(observeState)
  26.       return { unsubscribe }
  27.     },
  29.     [$$observable]() {
  30.       return this
  31.     }
  32.   }
  33. }
  36. function subscribe(listener: () => void) {
  37.   if (typeof listener !== 'function') {
  38.     throw new Error('Expected the listener to be a function.')
  39.   }
  41.   if (isDispatching) {
  42.     throw new Error(
  43.       'You may not call store.subscribe() while the reducer is executing. ' +
  44.       'If you would like to be notified after the store has been updated, subscribe from a ' +
  45.       'component and invoke store.getState() in the callback to access the latest state. ' +
  46.       'See https://redux.js.org/api/store#subscribelistener for more details.'
  47.     )
  48.   }
  50.   let isSubscribed = true
  52.   ensureCanMutateNextListeners()
  53.   nextListeners.push(listener)
  55.   return function unsubscribe() {
  56.     if (!isSubscribed) {
  57.       return
  58.     }
  60.     if (isDispatching) {
  61.       throw new Error(
  62.         'You may not unsubscribe from a store listener while the reducer is executing. ' +
  63.         'See https://redux.js.org/api/store#subscribelistener for more details.'
  64.       )
  65.     }
  67.     isSubscribed = false
  69.     ensureCanMutateNextListeners()
  70.     const index = nextListeners.indexOf(listener)
  71.     nextListeners.splice(index, 1)
  72.     currentListeners = null
  73.   }
  74. }

设计模式

  • 观察者模式
  • 迭代器模式

实现

  1. class Observable {
  2.   constructor(_subscribe) {
  3.     this._subscribe = _subscribe;
  4.   }
  6.   subscribe(observer) {
  7.     const safeObserver = new SafeObserver(observer);
  8.     safeObserver.unsub = this._subscribe(safeObserver);
  9.     return safeObserver.unsubscribe.bind(safeObserver);
  10.   }
  11. }
  13. Observable.prototype.map = function (project) {
  14.   return new Observable((observer) => {
  15.     const mapObserver = {
  16.       next: (x) => observer.next(project(x)),
  17.       error: (err) => observer.error(err),
  18.       complete: () => observer.complete()
  19.     };
  20.     return this.subscribe(mapObserver);
  21.   });
  22. };
  1. class SafeObserver {
  2.   constructor(destination) {
  3.     if (typeof destination !== 'object' || destination === null) {
  4.       throw new TypeError('Expected the observer to be an object.')
  5.     }
  7.     this.destination = destination;
  8.   }
  10.   next(value) {
  11.     // 尚未取消订阅，且包含next方法
  12.     if (!this.isUnsubscribed && this.destination.next) {
  13.       try {
  14.         this.destination.next(value);
  15.       } catch (err) {
  16.         // 出现异常时，取消订阅释放资源，再抛出异常
  17.         this.unsubscribe();
  18.         throw err;
  19.       }
  20.     }
  21.   }
  23.   error(err) {
  24.     // 尚未取消订阅，且包含error方法
  25.     if (!this.isUnsubscribed && this.destination.error) {
  26.       try {
  27.         this.destination.error(err);
  28.       } catch (e2) {
  29.         // 出现异常时，取消订阅释放资源，再抛出异常
  30.         this.unsubscribe();
  31.         throw e2;
  32.       }
  33.       this.unsubscribe();
  34.     }
  35.   }
  37.   complete() {
  38.     // 尚未取消订阅，且包含complete方法
  39.     if (!this.isUnsubscribed && this.destination.complete) {
  40.       try {
  41.         this.destination.complete();
  42.       } catch (err) {
  43.         // 出现异常时，取消订阅释放资源，再抛出异常
  44.         this.unsubscribe();
  45.         throw err;
  46.       }
  47.       this.unsubscribe();
  48.     }
  49.   }
  51.   unsubscribe() { // 用于取消订阅
  52.     this.isUnsubscribed = true;
  53.     if (this.unsub) {
  54.       this.unsub();
  55.     }
  56.   }
  57. }

Rx.js 响应式编程