How to merge "static" observable with "dynamic" in rxjs 6?

Question

I'm writing chrome extension and decided to wrap everything into streams, mostly for training and take a look into rxjs.

So I’m stuck with such a problem. I have two streams, one stream is static (don’t know how to correctly name it) and one dynamic.

A static stream is that one who emits value only when you subscribe to it (getPath$), it return value from a chrome storage. A dynamic (message$) stream is a stream which listening for events(messages).

So my goal is somehow to merge this two stream and every time when I’m receiving a message from message$ get value from a getPath$ and make some calculations based on two values.

type Handler = (
    message: any, 
    sender: any, 
    sendResponse: (response: any) => void
) => void; 

type Values  = string | string[];

const getValues = (values: Values) => (cb) => chrome.storage.local.get(values, cb) 
const getPathBinded = bindCallback(getValues('path')) 
const getPath$ = getPathBinded() 

const message$ = fromEventPattern( 
   (handler: Handler) => chrome.runtime.onMessage.addListener(handler), 
   (handler: Handler) => chrome.runtime.onMessage.removeListener(handler), 
   (message, sender, sendResponse) => ({ message, sender, sendResponse }) 
).pipe( 
    map(
       ({ message }) => message
    ) 
)

I found how to do this in such a way:

message$.pipe( 
    switchMap( 
        _ => getPath$, 
        (oV, iV) => {
            //doing some calculation
        } 
    ) 
 )

Or

combineLatest(
  message$,
  getPath$,
).pipe(
    map((a, b) => {
       //doing some calculation
    })
)

Seems like it's working, but it feels like I am doing wrong. Any suggestions or how to do it following best approaches? Also, please correct me in definitions.

UPD Here the full code: https://gist.github.com/ And the latest version which is work

const merged$ = message$.pipe(
   switchMap(sendedPath => combineLatest(
        path$,
        unit$
    ).pipe(
        map(([path, unit]) => [sendedPath, path, unit]))
    )
)

user_5699993 · Answer

If your goal is

to merge this two stream and every time when I’m receiving a message from message$ get value from a getPath$ and make some calculations based on two values

it looks like the switchMap approach is the right one. switchMap in this case means: any time I receive a notification from message$ I switch to the other Observable, in this case the one returned by invoking getPath$.

If you then want to do something with both values, then probably you can not afford to ignore the value notified by message$, which is what your code currently does, and you have to pipe a map into the result of getPath$ so that you can create an object containing the 2 values notified by the 2 Observables and pass such object to the following calculation. The code should look like

message$.pipe( 
    switchMap( 
        message => getPath$.pipe(map(path => ({message, path}))), 
        map(({message, path}) => {
            //doing some calculation
        })
    ) 
 )

user_3772379 · Answer

What you name dynamic/static is similar to Hot and Cold Observables, but not exactly the same.

Your first approach with switchMap is mostly fine, except that you should recalculate it every time:

const combined$ = messages$.pipe(
  mergeMap(() => getPathBinded(), (message, path) => { ... })
)

I have also changed switchMap to mergeMap, otherwise, you may "lost" some of the messages, see RxJS: Avoiding switchMap-Related Bugs

And also you may simplify your code a little bit:

const getStorage = bindCallback(chrome.storage.local.get); 
// use it as: const path$ = getStorage("path")

const message$ = fromEvent(chrome.runtime.onMessage);    
// should be enough, since you are only using default methods

How to merge "static" observable with "dynamic" in rxjs 6?

2 Answers