我正在尝试编写一个基于 websocket rxjs 的包装器。
而且我正在为我的 rxjs 理解而苦苦挣扎。
我有一个暂停流，它应该在发生错误时暂停可暂停的缓冲流，并在我从 websocket 获得“ok”后恢复它们。
不知何故，只有我的可暂停缓冲流上的第一个订阅被触发。从那时起，只有队列堆积得更高。
我已经准备了一个 jsbin 来重现这个问题。
https://jsbin.com/mafakar/edit?js,console
只有第一次订阅才会触发“msg recived”流。然后 q 和观察者开始叠加。
不知何故，我觉得这是关于冷热可观察的，但我无法理解这些问题。我将不胜感激。
提前谢谢你！
【问题讨论】：
标签： javascript rxjs
这不是冷/热问题。您在 onMessage 中所做的是订阅，然后处置。 dispose 终止序列。 onMessageStream 应该只订阅一次，例如在构造函数中：
this.onmessageStream.subscribe(message => console.log('--- msg --- ', message.data));
应删除订阅块，包括处置。
另外，请注意，您使用的 replaySubject 没有计数，这意味着队列包含所有先前的值。除非这是一种期望的行为，否则考虑将其更改为 .replaySubject(1)
【讨论】：
正如@Meir 指出的那样，订阅块中的dispose 是不可以的，因为它的行为是不确定的。一般来说，我会避免使用Subjects，而是使用工厂方法。您可以在此处查看重构版本：https://jsbin.com/popixoqafe/1/edit?js,console
更改的快速细分：
class WebSocketWrapper {
// Inject the pauser from the external state
constructor(pauser) {
// Input you need to keep as a subject
this.input$ = new Rx.Subject();
// Create the socket
this._socket = this._connect();
// Create a stream for the open event
this.open$ = Rx.Observable.fromEvent(this._socket, 'open');
// This concats the external pauser with the
// open event. The result is a pauser that won't unpause until
// the socket is open.
this.pauser$ = Rx.Observable.concat(
this.open$.take(1).map(true)
pauser || Rx.Observable.empty()
)
.startWith(false);
// subscribe and buffer the input
this.input$
.pausableBuffered(this.pauser$)
.subscribe(msg => this._socket.send(msg));
// Create a stream around the message event
this.message$ = Rx.Observable.fromEvent(this._socket, 'message')
// Buffer the messages
.pausableBuffered(this.pauser$)
// Create a shared version of the stream and always replay the last
// value to new subscribers
.shareReplay(1);
}
send(request) {
// Push to input
this.input$.onNext(request);
}
_connect() {
return new WebSocket('wss://echo.websocket.org');
}
}
顺便说一句，您还应该避免依赖像source 这样的内部变量，这些变量不适合外部使用。尽管 RxJS 4 相对稳定，但由于它们不是供公众使用的，它们可以在你的领导下进行更改。
【讨论】：