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

Question

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

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

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

所以我的目标是以某种方式合并这两个流，并且每次当我收到来自 message$ 的消息时，都从 getPath$ 获取值并进行一些计算基于两个值。

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

如果你的目标是

合并这两个流，每次我收到来自的消息时 message$ 从 getPath$ 获取值并根据两个值进行一些计算

看起来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
        })
    ) 
 )

Answer 2

你所说的动态/静态类似于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