Is there a way to address the sporadic behavior of rxjs combineLatest when used in conjunction with ReplaySubject

Question

Is there a way to address the sporadic behavior of rxjs combineLatest when used in conjunction with ReplaySubject

My current struggle lies within this particular example:

const r1 = new ReplaySubject(2);
const r2 = new ReplaySubject(2);
r1.next('r1.1');
r1.next('r1.2');
r2.next('r2.1');

combineLatest([r1, r2]).subscribe(console.log);  // -> ["r1.2", "r2.1"]
combineLatest([r2, r1]).subscribe(console.log);  // -> ["r2.1", "r1.1"] -> ["r2.1", "r1.2"]

// r1.1 ----> r1.2 --------
// -------------------> r2.1

After much contemplation, I find that my understanding may be hindered by the application's execution rather than the function's given specifications to produce the latest combinations.

Nevertheless, my quest is for a functionality similar to "combineEvery" that consistently outputs the second value. How might I go about implementing this?

// r1.1 ----> r1.2   
// -------------------> r2.1
// -------------------------------> r3.1

[r1.1, r2.1, r3.1] -> [r1.2, r2.1, r3.1]

javascript typescript rxjs

Answer 1

Answer №1

The behavior of the code snippet above is a result of how the combineLatest function operates. It waits to emit values until it has received at least one value from each observable. This is achieved by keeping track of the initial emissions from the inner observables and only emitting values once the counter matches the total count of all observables.

When using ReplaySubject, only the last element in the array of ReplaySubject instances will have all its buffered values emitted immediately. This is because the emissions from the other instances occur before the counter reaches the total count of all observables.

const r1 = new ReplaySubject(3);
const r2 = new ReplaySubject(3);
const r3 = new ReplaySubject(3);
r1.next('r1.1');
r1.next('r1.2');
r1.next('r1.3');
r2.next('r2.1');
r2.next('r2.2');
r3.next('r3.1');
combineLatest([r1, r2, r3]).subscribe(console.log);  // -> ["r1.3", "r2.2", "r3.1"]

To achieve the desired outcome, one approach is to prime the values with the first result and then introduce an artificial delay using the timer operator. Subscribing to the remaining emissions after this setup allows for the retrieval of the expected values.

combineLatest([
  concat(r1.pipe(first()), timer(0).pipe(switchMap(() => r1.pipe(skip(1)))),
  concat(r2.pipe(first()), timer(0).pipe(switchMap(() => r2.pipe(skip(1)))),
  concat(r3.pipe(first()), timer(0).pipe(switchMap(() => r3.pipe(skip(1))))
]).subscribe(console.log);  
// Using the values from the previous example, the following output is expected:
// ["r1.1", "r2.1", "r3.1"]
// ["r1.2", "r2.1", "r3.1"]
// ["r1.3", "r2.1", "r3.1"]
// ["r1.3", "r2.2", "r3.1"]

Answer 2

The behavior of the code snippet above is a result of how the combineLatest function operates. It waits to emit values until it has received at least one value from each observable. This is achieved by keeping track of the initial emissions from the inner observables and only emitting values once the counter matches the total count of all observables.

When using ReplaySubject, only the last element in the array of ReplaySubject instances will have all its buffered values emitted immediately. This is because the emissions from the other instances occur before the counter reaches the total count of all observables.

const r1 = new ReplaySubject(3);
const r2 = new ReplaySubject(3);
const r3 = new ReplaySubject(3);
r1.next('r1.1');
r1.next('r1.2');
r1.next('r1.3');
r2.next('r2.1');
r2.next('r2.2');
r3.next('r3.1');
combineLatest([r1, r2, r3]).subscribe(console.log);  // -> ["r1.3", "r2.2", "r3.1"]

To achieve the desired outcome, one approach is to prime the values with the first result and then introduce an artificial delay using the timer operator. Subscribing to the remaining emissions after this setup allows for the retrieval of the expected values.

combineLatest([
  concat(r1.pipe(first()), timer(0).pipe(switchMap(() => r1.pipe(skip(1)))),
  concat(r2.pipe(first()), timer(0).pipe(switchMap(() => r2.pipe(skip(1)))),
  concat(r3.pipe(first()), timer(0).pipe(switchMap(() => r3.pipe(skip(1))))
]).subscribe(console.log);  
// Using the values from the previous example, the following output is expected:
// ["r1.1", "r2.1", "r3.1"]
// ["r1.2", "r2.1", "r3.1"]
// ["r1.3", "r2.1", "r3.1"]
// ["r1.3", "r2.2", "r3.1"]

Is there a way to address the sporadic behavior of rxjs combineLatest when used in conjunction with ReplaySubject

Answer №1

Similar questions

What is the best way to call a method of a JavaScript object after calling a different method?

Vue - Syntax error: Unexpected token, expecting "}."

Adding a combination of HTML and Javascript to an element can result in unexpected behavior

What is the method for submitting a form when a change occurs?

ng filtering with a controller-defined scope

A guide on efficiently inserting multiple rows containing JSON data into a MySQL database simultaneously using node.js

What could be the reason for the unavailability of this.state in this particular situation?

How can I extract the text from a textarea in react-bootstrap and pass it to a function when a button is clicked?

Encountering issue while resolving flux/utils in webpack

Communicate through PHP and JavaScript chat to display HTML content in the chat window

JavaScript Game Timer

Create a custom CSS style to replace the default jQuery hide() function

Adding data to a defaultContent JSON column in JQuery DataTable using JavaScript

What is the best way to extract the value from a resolved Promise?

The issue with hiding and showing elements using JavaScript during drag and drop functionality

What is the proper method for utilizing an object as a dependency when using useEffect?

Update the dropdown menu based on the revised time

Are there any Android applications specifically designed for editing Typescript files?

Angular Pagination: A Detailed Guide

Quarterly Date Selection Tool with jQuery