Coding Challenge Practice - Question 108

The task is to implement the Subject Observable.

The boilerplate code

class Subject {
  constructor() {

  }
  subscribe(subscriber) {

  }
}

A Subject needs to keep a list of all active subscribers, allow new subscribers, notify all subscribers when next is called, support error and complete, and unsubscribe.

Store all active subscribers

this.observers = [];

Subscribe accepts a function or an observer object, adds it to observers and allows unsubscription.

subscribe(subscriber) {
    const observer =
      typeof subscriber === "function"
        ? { next: subscriber }
        : subscriber;

    if (this.isStopped) {
      return { unsubscribe() {} };
    }

    this.observers.push(observer);

    return {
      unsubscribe: () => {
        this.observers = this.observers.filter(
          (obs) => obs !== observer
        );
      }
    };
  }

next broadcasts the value to every subscriber

 next = (value)  => {
    if (this.isStopped) return;

    this.observers.forEach((observer) => {
      observer.next && observer.next(value);
    });
  }

error stops the Subject completely, notifies all subscribers, and clears the observers list

error = (err) => {
    if (this.isStopped) return;
    this.isStopped = true;

    this.observers.forEach((observer) => {
      observer.error && observer.error(err);
    });

    this.observers = [];
  }

complete works the same way as error

complete = () => {
    if (this.isStopped) return;
    this.isStopped = true;

    this.observers.forEach((observer) => {
      observer.complete && observer.complete();
    });

    this.observers = [];
  }

The final code

class Subject {
  constructor() {
    this.observers = [];
    this.isStopped = false;

  }
  subscribe(subscriber) {
    const observer = typeof subscriber === "function" ? { next: subscriber } : subscriber;
    if(this.isStopped) {
      return { unsubscribe() {}}
    }
    this.observers.push(observer);

    return {
      unsubscribe: () => {
        this.observers = this.observers.filter(
          (obs) => obs !== observer
        )
      }
    }
  }

  next = (value) => {
    if(this.isStopped) return;

    this.observers.forEach((observer) => {
      observer.next && observer.next(value);
    })
  }

  error = (err) => {
    if(this.isStopped) return;
    this.isStopped = true;

    this.observers.forEach((observer) => {
      observer.error && observer.error(err);
    })
    this.observers = [];
  }

  complete = () => {
    if(this.isStopped) return;
    this.isStopped = true;

    this.observers.forEach((observer) => {
      observer.complete && observer.complete()
    })
    this.observers = [];
  }
}

That's all folks!