时间戳

首页 > 解决方案 > 链接观察者 RxJs

typescript - 链接观察者 RxJs

问题描述

我正在使用 Observables 在我的 Angular 程序中实现 WebSocket 服务。到目前为止我有

    readwrite(commands: command[]) : Observable<response[]>{
        const observabe = new Observable((observer)=>{
            this.socket.subscribe((socket:WebSocket)=>{
                const id = this.uid++;
                //// Not Sure About This ////
                this.observerMap[id]={
                    next: next=>{
                        observer.next(next);
                        delete this.observerMap[id]; ////<---- What I want to Achieve
                    },
                    error: error=> observer.error(error),
                    complete: ()=>observer.complete()
                }
                socket.send(JSON.stringify({requestType:'ReadWrite', commands: commands, id: id}));
            });
        });
        return observable;
    }

然后在我的ws.onmessgae我有

    {
        const result = JSON.parse(event.data);
        this.observerMap[result.id]?.next(result.commands);
    }

这似乎符合我的要求，但我不确定是否有更清洁的方法来处理这个问题。我无法对 in 进行清理，observableMap[]因为ws.onmessage其他消息在多条消息中拥有一个观察者。我也只想在.next()处理完 Observable 之后进行清理，这导致我找到了这个解决方案。在观察者执行之前使用.pipe()导致执行。

有没有更简洁的方法将这些操作链接在一起？

标签： typescriptrxjsobservable

解决方案

我认为您应该通过以下方式使用完整的功能：

readwrite(commands: command[]) : Observable<response[]>{
    const observabe = new Observable((observer)=>{
        this.socket.subscribe((socket:WebSocket)=>{
            const id = this.uid++;
            this.observerMap[id]={
                next: next=>{
                    observer.next(next);
                },
                error: error=> observer.error(error),
                complete: ()=> { 
                   observer.complete()
                   delete this.observerMap[id];
                }
            }
            socket.send(JSON.stringify({requestType:'ReadWrite', commands: commands, id: id}));
        });
    });
    return observable;
}

然后在onmessage：

{
    const result = JSON.parse(event.data);
    this.observerMap[result.id]?.next(result.commands);
    this.observerMap[result.id]?.complete();
}

推荐阅读