如何在rxjs 6中合并“静态”可观察与“动态”？

Question

我正在编写chrome扩展，并决定将所有内容包装成流，主要用于培训并查看rxjs。

所以我遇到了这样的问题。我有两个流，一个流是静态的（不知道如何正确命名）和一个动态。

静态流是仅在您订阅它时发出价值的流（getPath $），它从chrome存储返回值。动态（消息$）流是监听事件（消息）的流。

所以我的目标是以某种方式合并这两个流，每次当我从getPath $的消息$ get值接收消息时，根据两个值进行一些计算。

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
    ) 
) 

我发现如何以这种方式做到这一点：

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

要么

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

好像它在工作，但感觉我做错了。任何建议或如何按照最佳方法做到这一点？另外，请在定义中纠正我。

UPD这里完整的code: https://gist.github.com/和最新版本的工作

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

Answer 1

你所说的动态/静态类似于Hot and Cold Observables，但不完全相同。

你使用switchMap的第一种方法大多都很好，除了你每次都应该重新计算它：

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

我也将switchMap改为mergeMap，否则，你可能会“丢失”一些消息，请参阅RxJS: Avoiding switchMap-Related Bugs

---

您也可以稍微简化一下代码：

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

Answer 2

如果你的目标是

合并这两个流，每当我从getPath $收到消息$ get value的消息时，根据两个值进行一些计算

看起来switchMap方法是正确的。 switchMap在这种情况下意味着：每当我收到message$的通知时，我都会切换到另一个Observable，在这种情况下是通过调用getPath$返回的。

如果你想对这两个值做一些事情，那么你可能无法忽略message$通知的值，这是你的代码当前所做的，你必须将map输入到getPath$的结果中，以便你可以创建包含2个Observables通知的2个值的对象，并将此对象传递给以下计算。代码看起来应该是这样的

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