Create a basic implementation of a streams API

Create a basic implementation of a streams API. The user should be able to push values to a stream, and subscribe to values that are pushed to that stream.

Example

Input:
const z = new Stream();
z.subscribe((value) => console.log(value));
z.subscribe((value) => console.log(value * 2));
z.subscribe((value) => console.log(value * 3));
z.push(2);

Output:
2
4
6

I think you would have got a basic idea about the question and what we have to exactly implement but still, let me walk you through it once.

We have to create a basic implementation of a streams API.

Streams API, pass the data in chunks which means as we keep pushing values it should go through the all the subscriptions.

To create this all we have to do is cache the subscriptions and whenever a value is pushed, call all the subscription methods with this value.

We will see two different implementations of these in Javascript.

1. Class-based.
2. Function-based.

Class-based implementation.

In the constructor we will initialize the cache and then create two methods inside the class subscribe and push.

subscribe will cache all the methods passed to it and we in the push method we will call all the subscription methods with the value received as an argument.

class Stream {
  constructor(){
    this.subscriptions = [];
  }
  
  subscribe(method){
    if(typeof method !== 'function' ){
      throw new Error('Invalid method!.');
    }
    
    this.subscriptions.push(method);
  }
  
  push(val){    
    this.subscriptions.forEach((method) => {
    	method.call(this, val);
    });
  }
}

Input:
const z = new Stream();
z.subscribe((value) => console.log(value));
z.subscribe((value) => console.log(value * 2));
z.subscribe((value) => console.log(value * 3));
z.push(2);

Output:
2
4
6

---

Function-based implementation.

The same can be achieved with function.

function Stream() {
  let subscriptions = [];
  
  this.subscribe = (method) => {
    if(typeof method !== 'function'){
      throw new Error('Invalid method!.');
    }
    
    subscriptions.push(method);
  }
  
  this.push = (val) => {    
    subscriptions.forEach((method) => {
      method.call(this, val);
    });
  }
}

Input:
const z = new Stream();
z.subscribe((value) => console.log(value));
z.subscribe((value) => console.log(value * 2));
z.subscribe((value) => console.log(value * 3));
z.push(2);

Output:
2
4
6

Please note, that the interview can ask you to create a different implementation of it, like an option to unsubscribe and pop values.