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

When the button is clicked, the JavaScript function is not being executed

CORS blocked in the live environment

Using Javascript, access the 'second child' element by referencing the 'first child' element within a shared 'parent element'

Narrowing down the keys of an indexable type in TypeScript based on their corresponding value types

The issue of basic authentication failing to function on Internet Explorer and Chrome, yet operating successfully on Firefox

What is the best way to extract a variable from a URL and assign it to an Ionic variable

Reveal or conceal elements with a touch of animation

Struggles with loading order in Knockout.JS

Import HTML document into a Bootstrap Popup

I'm curious if there is an eslint rule specifically designed to identify and flag any unnecessary spaces between a block comment and the function or

Look for identical values within a nested array

What steps should I take to incorporate Google sign-in on my website and gather user information efficiently?

Best practice for passing a variable argument in a JavaScript callback?

Setting up a function in React to utilize an image stored in state

Utilizing relative URIs in Node.js request library

Exploring the possibilities of combining colspan and ngFor in an Angular material 6 table

Trouble with Vue 3 watch not persisting after page refresh

Error message: Unable to load image from local source in Vue application

When creating a dynamic page number using JavaScript during a print event, the height of an A4 page is not taken into

JQuery not properly validating inputs with required attributes after creation