typescript - 如何在 rxjs 6 中将 "static"observable 与 "dynamic"合并？

Question

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

所以我被这样的问题困住了。我有两个流，一个流是static (不知道如何正确命名它)，一个是dynamic 。

静态流是仅当您订阅它时才发出值的流 (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

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