rewrite a rxjs class (Subject classes: Behavior, Replay, Async)

This is my experiment rewriting a part of rxjs lib with pure js, that provide same effect as the original one

I made a simple version of BehaviorSubject before (https://viblo.asia/p/rewrite-a-rxjs-class-behaviorsubject-6J3Zg3NBZmB).

Today I will implement others type of Subject including: BehaviorSubject, ReplaySubject, AsyncSubject using OOP. I guess I will try to make another one using FP as well.

For the test cases, pls check the original doc https://rxjs.dev/guide/subject

Full implementation is at https://codesandbox.io/s/rxjs-rewrite-subjects-u206x

A. Overall

These are all classes I created:

  1. A base class MySubject that have these methods:

    class MySubject {
    
      subscribe(obj) { //store subscriber's callbacks and provide unsubscribe function 
      }
    
      next(nextValue) { //push new value provided to value list 
      }
    
      error(e) {//manage subject status (stopped or not)  and trigger subscriber's callbacks accordingly
      }
    
      complete() {//same as error
      }
    }
    
  2. A 2nd base class MySyncSubject extends MySubject

    class MySyncSubject extends MySubject {
      next(nextValue) {// override next method to trigger subscriber's next callback
      }
    }
    
  3. MyReplaySubject extends MySyncSubject

    class MyReplaySubject extends MySyncSubject {
      subscribe(fn) {// override subscribe method to broadcast values based on bufferSize and timeFilter
      }
    }
    
  4. MyBehaviorSubject extends MyReplaySubject

    This is BehaviorSubject rewritted, a special one of MyReplaySubject which have bufferSize = 1 and no timeFilter

  5. MyAsyncSubject extends MySubject

    class MyAsyncSubject extends MySubject {
      complete() {//override complete method to broadcast the last value to all subscribers when subject completed
      }
    }
    

B. Details

  1. MySubject
    class MySubject {
      constructor() {
        this.isStopped = false;
        this.subscribers = [];
        this.value = [];
      }
    
      subscribe(obj) {//store subscriber's callbacks and provide unsubscribe function
        this.subscribers.push(obj);
        return () => {
          this.subscribers.pop();
        };
      }
    
      next(nextValue) {//push new value provided to value list
        if (!this.isStopped) {
          this.value.push({ nextValue: nextValue, timestamp: Date.now() });
        }
      }
      error(e) {//manage subject status (stopped or not)  and trigger subscriber's callbacks
        if (!this.isStopped) {
          this.isStopped = true;
          this.subscribers.forEach((obj) => obj.error(e));
        }
      }
    
      complete() {
        if (!this.isStopped) {
          this.isStopped = true;
          this.subscribers.forEach((obj) => obj.complete());
        }
      }
    }
    
  2. MySyncSubject
    class MySyncSubject extends MySubject {
     next(nextValue) {// override next method to trigger subscriber's next callback
       super.next(nextValue);
       if (!this.isStopped) {
         this.subscribers.forEach((obj) => obj.next(nextValue));
       }
     }
    }
    
  3. MyReplaySubject
    class MyReplaySubject extends MySyncSubject {
     constructor(bufferSize, timeLimit = -1) {
       super();
       this.bufferSize = bufferSize;
       this.timeFilter = (i) => {
         return timeLimit > 0 ? Date.now() - i.timestamp < timeLimit : true;
       };
     }
     subscribe(fn) {// override subscribe method to broadcast values based on bufferSize and timeFilter
       this.value
         .slice(Math.max(this.value.length - this.bufferSize, 0))
         .filter(this.timeFilter)
         .map((i) => fn.next(i.nextValue));
    
       super.subscribe(fn);
     }
    }
    
  4. MyBehaviorSubject
    class MyBehaviorSubject extends MyReplaySubject {
     constructor(defaultValue) {// override bufferSize = 1 and no timeFilter
       super();
       this.value = [{ nextValue: defaultValue, timestamp: Date.now() }];
       this.bufferSize = 1;
       this.timeFilter = () => true;
     }
    }
    
  5. MyAsyncSubject
    complete() {//override ```complete``` method to broadcast the last value to all subscribers when subject completed
       super.complete();
       this.subscribers.forEach((obj) =>
         obj.next(this.value[this.value.length - 1].nextValue)
       );
     }
    

All Rights Reserved