Merge topics together in RxJS like zip

Question

Merge topics together in RxJS like zip

Is it possible to create an observable that combines two subjects in a unique way, different from the zip function?

The goal is to combine two subjects so that when both have emitted values, the latest of their values is emitted. Then, after both emit at least once more, the latest values are emitted again, and so on.

Visual representation:

1 ---------- 2 ----- 3 -- 4 ------------------- 5 ——

------- a ------------------- b --- c — d —-————

Desired output:

------- 1a ----------------- 4b ---————-5d —-

For example, if subject1 emits 5 times and then subject2 emits once, the first emitted value would be a pair of both subjects' initial emissions: (subj1-emit5, subj2-emit1).

javascript angular typescript rxjs

Answer 1

Answer №1

If you want to accomplish your goal easily, consider using the `combineLatest()` function along with `take(1)` and then add the `repeat()` operator to resubscribe after each sequence:

import { Subject, combineLatest, take, repeat } from 'rxjs';

const s1$ = new Subject<number>();
const s2$ = new Subject<string>();

combineLatest([s1$, s2$]).pipe(
  take(1),
  repeat(),
).subscribe(console.log);

setTimeout(() => s1$.next(1), 0);
setTimeout(() => s2$.next('a'), 200);
setTimeout(() => s1$.next(2), 300);
setTimeout(() => s1$.next(3), 400);
setTimeout(() => s1$.next(4), 500);
setTimeout(() => s2$.next('b'), 600);
setTimeout(() => s2$.next('c'), 700);
setTimeout(() => s2$.next('d'), 800);
setTimeout(() => s1$.next(5), 900);

// Output: [1, 'a'], [4, 'b'], [5, 'd']

Check out the live demo here: https://stackblitz.com/edit/rxjs-4vg6lu?devtoolsheight=60&file=index.ts

Answer 2

If you want to accomplish your goal easily, consider using the `combineLatest()` function along with `take(1)` and then add the `repeat()` operator to resubscribe after each sequence:

import { Subject, combineLatest, take, repeat } from 'rxjs';

const s1$ = new Subject<number>();
const s2$ = new Subject<string>();

combineLatest([s1$, s2$]).pipe(
  take(1),
  repeat(),
).subscribe(console.log);

setTimeout(() => s1$.next(1), 0);
setTimeout(() => s2$.next('a'), 200);
setTimeout(() => s1$.next(2), 300);
setTimeout(() => s1$.next(3), 400);
setTimeout(() => s1$.next(4), 500);
setTimeout(() => s2$.next('b'), 600);
setTimeout(() => s2$.next('c'), 700);
setTimeout(() => s2$.next('d'), 800);
setTimeout(() => s1$.next(5), 900);

// Output: [1, 'a'], [4, 'b'], [5, 'd']

Check out the live demo here: https://stackblitz.com/edit/rxjs-4vg6lu?devtoolsheight=60&file=index.ts

Answer 3

Answer №2

To start creating an almostZip function, I would approach it like this (detailed comments are included within the code to explain the process)

function almostZip(o1$: Observable<any>, o2$: Observable<any>) {
  // A variable is set in the closure to maintain the state of the Observable
  // Specifically, it stores the values emitted by the 2 input Observables
  const valuesEmitted = [null, null]
  const _o1$ = o1$.pipe(
    // When o1$ emits, the emitted value is stored in the first position of the state variable
    tap(v => valuesEmitted[0] = v)
  )
  const _o2$ = o2$.pipe(
    // When o2$ emits, the emitted value is stored in the second position of the state variable
    tap(v => valuesEmitted[1] = v)
  )

  // This is the Observable returned by almostZip function
  return merge(_o1$, _o2$).pipe(
    // Filter out all notifications until both slots of the state var are not null
    // If both are not null, then we can notify the pair - as explained further down
    filter(() => {
      return !!valuesEmitted[0] && !!valuesEmitted[1]
    }),
    // Notify a copy of the state
    // Emitting a copy of the state is necessary to reset the state soon after
    map(() => [...valuesEmitted]),
    // Reset the state
    // After notifying a pair of values from each Observable, the state needs to be reset
    // to prepare for the next pair of values
    tap(() => {
      valuesEmitted[0] = null
      valuesEmitted[1] = null
    }),
  )
}

You can see a demonstration of how this functions on this stackblitz link

Answer 4

To start creating an almostZip function, I would approach it like this (detailed comments are included within the code to explain the process)

function almostZip(o1$: Observable<any>, o2$: Observable<any>) {
  // A variable is set in the closure to maintain the state of the Observable
  // Specifically, it stores the values emitted by the 2 input Observables
  const valuesEmitted = [null, null]
  const _o1$ = o1$.pipe(
    // When o1$ emits, the emitted value is stored in the first position of the state variable
    tap(v => valuesEmitted[0] = v)
  )
  const _o2$ = o2$.pipe(
    // When o2$ emits, the emitted value is stored in the second position of the state variable
    tap(v => valuesEmitted[1] = v)
  )

  // This is the Observable returned by almostZip function
  return merge(_o1$, _o2$).pipe(
    // Filter out all notifications until both slots of the state var are not null
    // If both are not null, then we can notify the pair - as explained further down
    filter(() => {
      return !!valuesEmitted[0] && !!valuesEmitted[1]
    }),
    // Notify a copy of the state
    // Emitting a copy of the state is necessary to reset the state soon after
    map(() => [...valuesEmitted]),
    // Reset the state
    // After notifying a pair of values from each Observable, the state needs to be reset
    // to prepare for the next pair of values
    tap(() => {
      valuesEmitted[0] = null
      valuesEmitted[1] = null
    }),
  )
}

You can see a demonstration of how this functions on this stackblitz link

Answer 5

Answer №3

What if we use the withLatestFrom operator?

import './custom-style.css';

import { rx, of, tap, interval, withLatestFrom } from 'rxjs';
const stream1$ = interval(1000);
const stream2$ = interval(5000);

stream2$
  .pipe(
    withLatestFrom(stream1$),
    tap(([firstData, secondData]) => {
      console.log(`Data from second source (every 5 seconds): ${firstData}, Data from first source (every 1 second): ${secondData}`);
    })
  )
  .subscribe();

// Check the browser console for results.

Try it on StackBlitz

Answer 6

What if we use the withLatestFrom operator?

import './custom-style.css';

import { rx, of, tap, interval, withLatestFrom } from 'rxjs';
const stream1$ = interval(1000);
const stream2$ = interval(5000);

stream2$
  .pipe(
    withLatestFrom(stream1$),
    tap(([firstData, secondData]) => {
      console.log(`Data from second source (every 5 seconds): ${firstData}, Data from first source (every 1 second): ${secondData}`);
    })
  )
  .subscribe();

// Check the browser console for results.

Try it on StackBlitz

Merge topics together in RxJS like zip

Answer №1

Answer №2

Answer №3

Similar questions

How can a JavaScript function be triggered by Flask without relying on any requests from the client-side?

React - Refreshing a component with the help of another component

Tips for adjusting image size to take up only half of the screen in NextJS

Population of the global namespace in JavaScript without the need for the 'new'

I am working on two Angular projects and I am looking to integrate one of them into the other project

Is there a way for me to manually manipulate the advancement of the progress bar from @ngx-progressbar/core in Angular5/Ionic4?

Creating a PHP script that retrieves data from JavaScript and stores it in MySQL can be accomplished by using AJAX to send the

Guide on applying a filter to the items in a listbox using the input from a text box

Javascript code to verify whether the page is currently positioned at the top

The Angular Material Table is reporting an error: the data source provided does not conform to an array, Observable, or DataSource

What is the best way to store an image file using html/angularjs?

The deployment of my Node application on Heroku is causing an error message: node-waf is not

Discover the magic of TransformToggle and slideToggle in Javascript, HTML, and CSS

I must interact with the video within the iframe by clicking on it

Flask caches JSON files automatically

What is the process for sending a message from the internet to a mobile device?

Uh oh! We encountered an error: Uncaught (in promise): Error: No routes found for the provided URL segment

Please ensure that the table contains all the records corresponding to the respective days

Converting ed25519 private key into OpenSSH Format using JavaScript

Discovering the quantity of items with a specific value in Angular 8