RXJS Observable stretch

user_1822033 0
Viewed 237

I have a Rx.Observable.webSocket Subject. My server endpoint can not handle messages receiving the same time (<25ms). Now I need a way to stretch the next() calls of my websocket subject.

I have created another Subject requestSubject and subscribe to this. Then calling next of the websocket inside the subscription.

requestSubject.delay(1000).subscribe((request) => {
  console.log(`SENDING: ${JSON.stringify(request)}`);
  socketServer.next(JSON.stringify(request));
});

Using delay shifts each next call the same delay time, then all next calls emit the same time later ... thats not what I want.

I tried delay, throttle, debounce but it does not fit.

The following should illustrate my problem

Stream 1 | ---1-------2-3-4-5---------6----

    after some operation ...

Stream 2 | ---1-------2----3----4----5----6-
rxjs reactive-programming javascript
3 Answers

Mark van Straten's solution didn't work completely accurately for me. I found a much more simple and accurate solution based from here.

const source = from([100,500,1500,1501,1502,1503]).pipe(
    mergeMap(i => of(i).pipe(delay(i)))
);

const delayMs = 500;
const stretchedSource = source.pipe(
  concatMap(e => concat(of(e), EMPTY.pipe(delay(delayMs))))
);
user_55591250
Related