What techniques can I employ to ensure that withLatestFrom() effectively interacts with itself?

Question

What techniques can I employ to ensure that withLatestFrom() effectively interacts with itself?

In my program, I have an intermediate stream that is connected to a source, but it can also trigger events from other sources (such as user input). In another part of my code, there is a derived stream that needs to compare new data from the intermediate with the last value of the source. This is the snippet of code that handles this scenario:

const source = new Rx.Subject;
const derived = new Rx.Subject;
derived.subscribe( () => console.log( "derived" ) );
const intermediate = new Rx.Subject;

// The reason for having the "intermediate" stream is that sometimes it triggers on its own:
source.subscribe( intermediate );

intermediate
    .withLatestFrom( source )
    .subscribe( derived );

source.next();

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.4.3/Rx.min.js"></script>

The issue here is that the "derived" message is never displayed (the first event in the source is being ignored). How can I create a stream that retrieves the last value of the source for every message from the intermediate stream, even if the source is currently propagating?

javascript rxjs

Answer 1

Answer №1

From what I gather, there is a flow involving a source stream, followed by an intermediate stream that subscribes to and emits all items from source along with additional ones from user input. Then there are two distinct scenarios at play, each requiring its own approach:

Outputting values of source with timings based on intermediate: Utilizing the combineLatest operator can easily achieve this task:

const source_with_timing_from_intermediate : Observable<TSource> = intermediate.combineLatest(source, (i, s) => s);

Comparing the most recent value from intermediate with source: Although it appears similar, using combineLatest may not be reliable because if you perform a simple operation like:

const is_different : Observable<bool> = intermediate.combineLatest(source, (i, s) => i !== s);

If source emits a new value before intermediate catches up, you might mistakenly attribute that previous value to intermediate.

For greater reliability, consider buffering and employing a subject:

// untested
const derived : Observable<TSource> = (function() {
  const source_buffer = new Queue();
  const subj = new Subject();
  source.forEach(s => {
    source_buffer.enqueue(s);
  });
  intermediate.forEach(i => {
    subj.onNext(source_queue.peek() === i);
    if(source_buffer.peek() === i) source_buffer.dequeue();
  });
  Promise.all([
    source.toPromise(),
    intermediate.toPromise()
  ]).then(() => subj.onClose());
  return subj.asObservable();
})();

Answer 2

From what I gather, there is a flow involving a source stream, followed by an intermediate stream that subscribes to and emits all items from source along with additional ones from user input. Then there are two distinct scenarios at play, each requiring its own approach:

Outputting values of source with timings based on intermediate: Utilizing the combineLatest operator can easily achieve this task:

const source_with_timing_from_intermediate : Observable<TSource> = intermediate.combineLatest(source, (i, s) => s);

Comparing the most recent value from intermediate with source: Although it appears similar, using combineLatest may not be reliable because if you perform a simple operation like:

const is_different : Observable<bool> = intermediate.combineLatest(source, (i, s) => i !== s);

If source emits a new value before intermediate catches up, you might mistakenly attribute that previous value to intermediate.

For greater reliability, consider buffering and employing a subject:

// untested
const derived : Observable<TSource> = (function() {
  const source_buffer = new Queue();
  const subj = new Subject();
  source.forEach(s => {
    source_buffer.enqueue(s);
  });
  intermediate.forEach(i => {
    subj.onNext(source_queue.peek() === i);
    if(source_buffer.peek() === i) source_buffer.dequeue();
  });
  Promise.all([
    source.toPromise(),
    intermediate.toPromise()
  ]).then(() => subj.onClose());
  return subj.asObservable();
})();

What techniques can I employ to ensure that withLatestFrom() effectively interacts with itself?

Answer №1

Similar questions

Utilizing a drop-down selection menu and a designated container to store chosen preferences

Can anyone provide guidance on how to simulate a click on a JavaScript action for an iPhone?

Is there a way to display the JavaScript widget exclusively on the homepage or main page

Angular: implementing a service for conditional module imports

Maintaining datatype integrity in MongoDB Stitch when decrementing with update statements

How can I optimize the performance of JS-heavy pages on mobile devices?

What is the best way to access an item within an item using HTML and AngularJS?

Iterate through an array index within a map function in a ReactJS component

Issue: jQuery Datatable not functioning properly while attempting to populate the table through JavaScript.Explanation: The

The behavior of Datatables varies depending on the screen resolution

Unforeseen SyntaxError: Unexpected symbol detected

Determine if a mobile application has been installed using Vue.js

What is the best way to anticipate a return phone call?

Display additional inputs using the PHP Foreach Loop depending on the selection made

Issue: ENOENT - The specified file or directory cannot be found while scanning in a discord bot

Can we show the dropdown with the active button?

handle an exception within the initializer of its object

Next.js 14 useEffect firing twice upon page load

The novice image slideshow script in JavaScript is causing all images to disappear and generating errors

The clingy navigation bar hinders smooth scrolling to the content