Nrthugu

I'm playing with RxJS for the first time, and I'm trying to create a simple observable that will take both a maximum number of entries and wait for a time interval to pass before passing the data on to subscribers.

This is what I have:

import EventEmitter from "events";



import { fromEvent } from "rxjs";

import { bufferTime, filter, take } from "rxjs/operators";



const ev = new EventEmitter();

const observer = fromEvent(ev, "log").pipe(

  take(10),

  bufferTime(1000),

  filter(logs => !!logs.length)

);



observer.subscribe(data => {

  console.log(`received: ${data.length}`);

});



for (let i = 0; i < 100; i++) {

  ev.emit("log", { hello: "world" });

}

What I expected to happen:

received: 10 to be printed 10x, at 1 second intervals.

What actually happened:

received: 10 was printed once, and then the script exits.

Why I think it's happening

I'm new to RxJS, but from what I can gather the take() operator emits a 'completed' status after it has taken 10 entries, which prevents any further subscriptions from firing.

How do I make this observable 'recurring' so that it'll a) take a maximum of 10, b) ensure it runs, at most, once every 1000ms, and c) repeat forever?

Try to insert a simple tap(d => console.log(JSON.stringify(d))), operator before take(10) and after bufferTime and you see what happens.

Basically you have a loop that emits 100 events. Out of this loop you create an Observable which notifies 100 times the object emitted by you event loop.

This whole thing happens synchronously in your example.

Then the Observable is transformed by a pipe which does the following:

• first takes just the first 10 notifications via the take operator


• then says "accumulate all the notifications which happen in the first
  1 second and emit them in form of an array" - considering that the
  whole emission process is synchronous and that you take the first 10
  of them, then bufferTime(1000) will emit one array of 10 items
  after 1 second


• nothing else will be emitted by this stream, since take(10)
  completes the source Observable after 10 synchronous notifications,
  so the filter operator is of no use


• last, in the subscription you receive just 1 notification, i.e. the only one emitted by bufferTime
  

Said that, probably what you want to achieve can be accomplished by code along these lines

const ev = new EventEmitter();



function simpleObservable(maxEntries: number, waitTime: number) {

  return fromEvent(ev, "log").pipe(

    bufferCount(maxEntries),

    filter(logs => !!logs.length),

    mergeMap(logs => of(logs).pipe(delay(waitTime)), 1),

  )

}



simpleObservable(10, 1000)

.subscribe(data => {

  console.log(`received: ${data.length}`);

});



for (let i = 0; i < 100; i++) {

  ev.emit("log", { hello: "world" });

}

The key here is that first you slice the stream of events into arrays of 10 events each, and this is done via the bufferCount operator.

Then you introduce the asynchronicity using mergeMap with concurrency level set to 1 (which is equivalent to the operator concatMap). You basically transform each array emitted by bufferCount into another Observable via the of function and apply a delay of 1 second to each of the new Observables. Then you concatenate them so that they are emitted each after the other with the time difference of 1 second.

UPDATED RESPONSE after @Lee Benson comment

bufferTime can be the answer to your problem.

bufferTime has 3 parameters:

• 
  bufferTimeSpan: specifies the lifespan of the buffer, i.e. the Observable emits the buffer and resets it after at each interval
  specified by bufferTimeSpan
  


• 
  bufferCreationInterval: to be left null for our purpose


• 
  maxBufferSize: specifies the max size of the buffer - if the size is reached, then the Observable emits the buffer and resets it

So, if we use the first and the last parameter we should be able to reach the desired objectives.

This is the test I have put together

const ev = new EventEmitter();

// I merge to the event stream created by 'ev' a single notification emitted after 5 seconds, 

// just to make the thing a bit more complex

const sourceObs = merge(

  fromEvent(ev, "log"),

  of(1).pipe(delay(5000))

)



function simpleObservable(maxEntries: number, waitTime: number) {

  return sourceObs.pipe(

    bufferTime(waitTime, null, maxEntries),

    filter(logs => !!logs.length),

  )

}



simpleObservable(10, 2000)

.subscribe(data => {

  console.log(`received: ${data.length}`);

});



for (let i = 0; i < 19; i++) {

  ev.emit("log", { hello: "world" });

}

// some other events are fired after 6 seconds, to make the source more complex

setTimeout(() => {

  for (let i = 0; i < 17; i++) {

    ev.emit("log", { hello: "world" });

  }

}, 6000);

After reading your last comment, a way to consider to solve the problem could be to combine bufferTime with interval using zip.

Basically the idea is that you can set a rhythm of notifications using interval - for instance you set interval(1000) to have an Observable that emits every second.

Then you can use bufferTime(1000, null, 10) to make sure that you emit an array every 1 second or if your buffer reaches 10 items, whichever comes first.

Now if you zip these 2 Observables you obtain an Observable which emits every second, because of interval(1000), and it emits also whatever comes out of bufferTime(1000, 0, 10) in sequence. So, if the source Observable emits more than 10 items in a second, the first 10 will be emitted in the first notification, the others will remain buffered in the Observable and will be emitted in the subsequent notifications in chunks of 10 every second.

The code probably makes this clearer.

const ev = new EventEmitter();

// I merge to the event stream created by 'ev' a single notification emitted after 5 seconds, 

// just to make the thing a bit more complex

const sourceObs = merge(

  fromEvent(ev, "log"),

  of(1).pipe(delay(5000))

)



function simpleObservable(maxEntries: number, waitTime: number) {

  return zip(

    timer(0, waitTime),

    sourceObs.pipe(

      bufferTime(waitTime, null, maxEntries),

    )

  )

  .pipe(

    filter(logs => !!logs[1].length),

    map(logs => logs[1])

  )

}



const maxEntries = 4;

const waitTime = 1000;

simpleObservable(maxEntries, waitTime)

.subscribe(data => {

  console.log(`received: ${data.length}`);

});



for (let i = 0; i < 19; i++) {

  ev.emit("log", { hello: "world " + i + 'A' });

}

// some other events are fired after 8 seconds, to make the source more complex

setTimeout(() => {

  for (let i = 0; i < 17; i++) {

    ev.emit("log", { hello: "world " + i + 'B' });

  }

}, 8000);

Clearly you need to consider that, if the source Observable emits at an speed higher than your ability to consume notifications, you may end up with memory problems.