rx.js

const logger = {
  next(value) {
    console.log(value);
  }
};
const logger2 = {
  next(value) {
    console.log("logger 2 " + value);
  }
};

class Observable {
  constructor(subscriberOps) {
    this.subscriptions = [];
    this.subscriberOps = subscriber => {
      if (!this.subscriptions.find(s => s.subscriber === subscriber)) {
        return;
      }
      subscriberOps(subscriber);
    };

    this.pipePlugins = [];
  }

  patchSubscriber(subscriber) {
    if (!subscriber.next) {
      return;
    }

    let decoratedSubscriber = subscriber;
    for (const plugin of this.pipePlugins) {
      decoratedSubscriber = plugin(decoratedSubscriber);
    }

    this.subscriptions.push({
      originSubscriber: subscriber,
      subscriber: decoratedSubscriber
    });

    return decoratedSubscriber;
  }

  dispatchSubscriber(subscriber) {
    if (!subscriber.next) {
      return;
    }

    this.subscriptions = this.subscriptions.filter(
      s => s.originSubscriber !== subscriber
    );
  }

  subscribe(subscriber) {
    let observableSubscriber = this.subscriptions.find(
      s => s.originSubscriber === subscriber
    );
    observableSubscriber =
      observableSubscriber && observableSubscriber.subscriber;
    if (!observableSubscriber) {
      observableSubscriber = this.patchSubscriber(subscriber);
    }
    this.subscriberOps(observableSubscriber);

    return this.unsubscriber(subscriber);
  }

  unsubscriber = subscriber => {
    return () => {
      if (this.subscriptions.find(s => s.originSubscriber === subscriber)) {
        this.dispatchSubscriber(subscriber);
      }
    };
  };

  unsubscribe = subscriber => {
    if (this.subscriptions.find(s => s.originSubscriber === subscriber)) {
      this.dispatchSubscriber(subscriber);
    }
  };

  pipe(...decorators) {
    for (const decorator of decorators) {
      this.pipePlugins.unshift(decorator);
    }
    return this;
  }
}

class Subject extends Observable {
  constructor() {
    super(subscriber => {
      if (!this.nextValue) {
        return;
      }

      subscriber.next(this.nextValue);
    });

    this.nextValue = undefined;
  }

  subscribe(subscriber) {
    let observableSubscriber = this.subscriptions.find(
      s => s.originSubscriber === subscriber
    );
    observableSubscriber =
      observableSubscriber && observableSubscriber.subscriber;
    if (!observableSubscriber) {
      observableSubscriber = this.patchSubscriber(subscriber);
    }

    return this.unsubscriber(subscriber);
  }

  next(value) {
    this.nextValue = value;
    for (const subscriber of this.subscriptions) {
      this.subscriberOps(subscriber.subscriber);
    }
  }
}

function echo() {
  return subscriber => ({
    next: value => {
      subscriber.next(value);
      subscriber.next(value);
    }
  });
}

function distinct() {
  const distinctValues = [];
  return subscriber => ({
    next: value => {
      if (distinctValues.includes(value)) {
        return;
      } else {
        distinctValues.push(value);
        subscriber.next(value);
      }
    }
  });
}

function map(mapRule) {
  return subscriber => ({
    next: value => {
      subscriber.next(mapRule(value));
    }
  });
}

function debounce(time) {
  return subscriber => ({
    next: value => {
      setTimeout(() => {
        subscriber.next(value);
      }, time);
    }
  });
}

function throttle(time) {
  let timeout;
  let stopped = true;
  let lastRunningValue;
  let currentValue;

  const stop = callback => setTimeout(callback, 10);

  return subscriber => ({
    next: value => {
      currentValue = value;
      if (timeout) {
        return;
      }

      if (stopped) {
        subscriber.next(currentValue);
        lastRunningValue = currentValue;
        stopped = false;
        return;
      }

      checkStop = true;
      timeout = setTimeout(() => {
        clearTimeout(timeout);
        timeout = undefined;
        lastRunningValue = currentValue;
        stop(() => {
          stopped = lastRunningValue === currentValue;
          subscriber.next(lastRunningValue);
        });
      }, time);
    }
  });
}

const o = new Observable(subscriber => {
  let count = 0;
  const display = document.getElementById("display");
  document.addEventListener("mousemove", () => {
    count++;
    display.innerText = count;
    subscriber.next(count);
  });
});

o.pipe(
  debounce(1000),
  throttle(1000),
  map(value => `throttle: ${value}`)
).subscribe(logger);

const s1 = new Subject();
s1.next("never");
s1.subscribe(logger);
s1.subscribe(logger2);
s1.next("hello world");
s1.unsubscribe(logger);
s1.next("hello world 2");
s1.subscribe(logger);
s1.next("hello world 3");

new Observable(subscriber => {
  [1, 2, 3, 4, 4, 3, 2, 1, 5].map(value => subscriber.next(value));
})
  .pipe(
    debounce(1000),
    distinct(),
    map(value => value * value),
    echo()
  )
  .subscribe(logger);