Merge rxjs streams, identify modifications, and yield a single result

Question

Merge rxjs streams, identify modifications, and yield a single result

In the context of using Angular with .net Core WebApi, let's consider the development of a time management application designed to monitor task durations. The user initiates a task on the front end which triggers a timer displaying elapsed time.

The concept involves retrieving the elapsed time data from the backend. When a user begins a task on the front end, a timer starts on the backend and sends the current elapsed time back to the front end for display. The goal is to ensure accurate time tracking even in cases of connection issues between the backend and frontend.

To illustrate this scenario: the backend stream sends out values every second (representing the elapsed time displayed to the user). However, if there is a temporary connection disruption after 6 seconds, causing a delay before sending "9" ("0:09"), it might confuse the user ("it was at 6 seconds and now suddenly 9?"). To address this, an interval is set up on the front end to emit a new value each second. This allows for checking whether the backend stream has sent a new value within that timeframe, and if not, the previous value can be adjusted to ensure the user sees the correct elapsed time.

bStream => ---1---2---3---4---5---6---x---x---9
fStream => ---1---2---3---4---5---6---7---8---9

What the user observes:
00:01 -> 00:02 -> 00:03 -> 00:04 -> 00:05 -> 00:06 (pause) -> 00:09

Desired user experience:
00:01 -> 00:02 -> 00:03 -> 00:04 -> 00:05 -> 00:06 -> (pause - frontend detects no new value and adds 1 second to the previous)
Resulting sequence:
00:01 -> 00:02 -> 00:03 -> 00:04 -> 00:05 -> 00:06 -> 00:07 -> 00:08 -> 00:09

Facing some challenges in determining where to begin this process.

Created a quick fiddle with two streams but struggling to figure out how to detect when bStream does not emit a new value.

https://stackblitz.com/edit/typescript-yookbj

javascript typescript .net-core rxjs ngrx

Answer 1

Answer №1

Here is a potential solution:

const be$ = concat(
  of(1).pipe(delay(100)),
  of(2).pipe(delay(100)),
  of(3).pipe(delay(100)),
  of(4).pipe(delay(100)),
  of(5).pipe(delay(100)),
  of(6).pipe(delay(100)),

  of(10).pipe(delay(500)), // Wait after freeze
  of(11).pipe(delay(100)),
  of(12).pipe(delay(100)),
).pipe(shareReplay({ bufferSize: 1, refCount: true, }), endWith(null));

// Skip(1) - the ReplaySubject used by shareReplay() will give us the latest value and it's not needed
const beReady$ = be$.pipe(skip(1), take(1));

const fe$ = be$.pipe(
  mergeMap(v => merge(
    of(v),
    of(v).pipe(
      expand(v => timer(100).pipe(map(v1 => 1 + v))),
      takeUntil(beReady$),
    )
  )),
  distinctUntilChanged(),
  filter(v => v !== null)
).subscribe(console.log)

endWith(null) - to halt the recursion when the final value (12) is emitted, we need the source to emit something else

shareReplay - the source must be shared as there will be an additional subscriber (beReady$), apart from the main subscriber (fe$)

mergeMap(v => merge(
  of(v), // Pass on the current value
  of(v).pipe(
    // If be$ does not emit within the next 100ms, send the currentValue + 1
    // and continue this process until be$ finally emits
    expand(v => timer(100).pipe(map(v1 => 1 + v))),
    takeUntil(beReady$),
  )
)),

expand works similarly to mergeMap, with some differences:

it passes along the inner value
it creates another inner observable based on the last inner value, making it recursive
takeUntil(beReady$) is how the recursion can be terminated

StackBlitz

Answer 2

Here is a potential solution:

const be$ = concat(
  of(1).pipe(delay(100)),
  of(2).pipe(delay(100)),
  of(3).pipe(delay(100)),
  of(4).pipe(delay(100)),
  of(5).pipe(delay(100)),
  of(6).pipe(delay(100)),

  of(10).pipe(delay(500)), // Wait after freeze
  of(11).pipe(delay(100)),
  of(12).pipe(delay(100)),
).pipe(shareReplay({ bufferSize: 1, refCount: true, }), endWith(null));

// Skip(1) - the ReplaySubject used by shareReplay() will give us the latest value and it's not needed
const beReady$ = be$.pipe(skip(1), take(1));

const fe$ = be$.pipe(
  mergeMap(v => merge(
    of(v),
    of(v).pipe(
      expand(v => timer(100).pipe(map(v1 => 1 + v))),
      takeUntil(beReady$),
    )
  )),
  distinctUntilChanged(),
  filter(v => v !== null)
).subscribe(console.log)

endWith(null) - to halt the recursion when the final value (12) is emitted, we need the source to emit something else

shareReplay - the source must be shared as there will be an additional subscriber (beReady$), apart from the main subscriber (fe$)

mergeMap(v => merge(
  of(v), // Pass on the current value
  of(v).pipe(
    // If be$ does not emit within the next 100ms, send the currentValue + 1
    // and continue this process until be$ finally emits
    expand(v => timer(100).pipe(map(v1 => 1 + v))),
    takeUntil(beReady$),
  )
)),

expand works similarly to mergeMap, with some differences:

it passes along the inner value
it creates another inner observable based on the last inner value, making it recursive
takeUntil(beReady$) is how the recursion can be terminated

StackBlitz

Answer 3

Answer №2

Kindly review the code snippet provided below. To simulate a synchronization issue, modify the code v * 5 to v * 4. This change will ensure that the counter aligns with the value obtained from the "backend" once it has been received.

// emit every 5s
const source = interval(5000).pipe(
    map(v1 => v1 + 1), 
    startWith(0), 

    map(v => v * 5),
);
// emit every 1s
const secondSource = interval(1000).pipe(
    delay(50), 
    map(v1 => v1 + 1), 
    startWith(0)
);

source.pipe( 
    switchMap(v => secondSource.pipe(
        map(v1 => v + v1), 
    )),
).subscribe(v => console.log(v));

The updated code now ensures smooth counting from 0 to N even in the presence of lags from the backend.

UPDATED

There is an even simpler method available, although it does not depend on the time when the backend responds and operates independently on a one-second period.

pipe(
  bufferTime(1000), 
  scan((a, b) => b.length ? a + 1 : b[0], 0), 
);

Answer 4

Kindly review the code snippet provided below. To simulate a synchronization issue, modify the code v * 5 to v * 4. This change will ensure that the counter aligns with the value obtained from the "backend" once it has been received.

// emit every 5s
const source = interval(5000).pipe(
    map(v1 => v1 + 1), 
    startWith(0), 

    map(v => v * 5),
);
// emit every 1s
const secondSource = interval(1000).pipe(
    delay(50), 
    map(v1 => v1 + 1), 
    startWith(0)
);

source.pipe( 
    switchMap(v => secondSource.pipe(
        map(v1 => v + v1), 
    )),
).subscribe(v => console.log(v));

The updated code now ensures smooth counting from 0 to N even in the presence of lags from the backend.

UPDATED

There is an even simpler method available, although it does not depend on the time when the backend responds and operates independently on a one-second period.

pipe(
  bufferTime(1000), 
  scan((a, b) => b.length ? a + 1 : b[0], 0), 
);

Merge rxjs streams, identify modifications, and yield a single result

Answer №1

Answer №2

Similar questions

The dropdown menu repeatedly opens the initial menu only

Utilize Angular Guards and RxJS to coordinate the flow of data between parent and child Guards. Ensure that the Child Guard waits for the parent

Tips for implementing UI properties in React

Can I create interactive stacked shapes with CSS and/or JavaScript?

It ceases to function when transferred to another file

We've encountered an issue with Redux and Typescript: 'Unable to find property 'prop' in type 'string[]'

Changing table data using a switch in mui-datatables when my information is stored as boolean values

A JavaScript function that smoothly scrolls an element into view while considering any scrollable or positioned parent elements

Retrieving the chosen value when there is a change in the select tag

"Customizing the template of the Angular Material 2 datepicker: A step-by-step

The then() function in Node.js is triggered before the promise is fully resolved

Set up specific vue.config.js configurations for unique environments in Vue

Updating a table without the need for a button in PHP and AJAX with an Excel-inspired approach

Retrieve information from arrays within objects in a nested structure

Having trouble with the DataTables jQuery plugin? Seeing a blank page instead of the expected results?

What is the best way to receive the information that was transmitted to me through a callback function?

Sending parameters within ajax success function

Utilizing React hooks to capture the checkbox value upon change and transfer it to the submitForm function

Plotting Polyline Path on Google Maps using Angular

"An issue arises when using req.body and res.render as the values retrieved are