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

Divide JSON arrays into separate select boxes

Toggling javascript functionality based on media queries

Pressing the button results in no action

What is the best way to eliminate all instances of the period symbol from an array?

Using Javascript function with ASP.NET MVC ActionLink

Continue looping in Javascript until an empty array is identified

Combine rows with the same value in the first column of an HTML table

There seems to be an issue with accessing /puffins/5f298d0ebcbaf254dcf282b3 at

Retrieve the URL for the React component

Can you create a stroke that is consistently the same width as the container BoxElement?

What steps can be taken to diagnose the cause of a failed Jquery AJAX request?

Failed PHP response when jQuery was called

What is the proper usage of a jwt token?

The MDX blog was set up to showcase markdown content by simply displaying it without rendering, thanks to the utilization of the MDXProvider from @mdx-js/react within Next JS

Trouble with Ajax program loading properly in Chrome browser

Tips for activating multiple CSS animations when scrolling

The unexpected occurence of the Onbeforeunload exception

I am interested in incorporating the ability to select and scroll the window without needing to interact with the scroll bar by

When you use map to transform the value, Observable will consistently return as undefined

Is it better to include the Google Analytics code in the master page or on every individual page of an asp.net