Angular Websocket RxJS / WebSocketSubject

Question

过去我们使用了几个 angular2 websocket impelmentations，但我们对它们并不满意，使用它们存在一些问题。所以我们决定用 RxJs 自己尝试一下。

这是我们的第一次尝试：

@Injectable()
export class WebSocketService{

    public createWebsocket(url: string): Subject<MessageEvent> {
        let socket = new WebSocket(url);

        let observable = Observable.create(
                                    (observer: Observer<MessageEvent>) => {
                                        socket.onmessage = observer.next.bind(observer);
                                        socket.onerror   = observer.error.bind(observer);
                                        socket.onclose   = observer.complete.bind(observer);

                                        return socket.close.bind(socket);
                                    });

        let observer = {
            next: (data: Object) => {
                if (socket.readyState === WebSocket.OPEN) {
                    socket.send(JSON.stringify(data));
                }
            }
        };

        return Subject.create(observer, observable);
    }
}

套接字已打开并正常工作了一段时间。几秒钟后，浏览器正在关闭套接字，我在服务器端收到一个关闭事件。

这是我们进入服务器站点的最终原因： [1006] WebSocket 读取 EOF

有人可以帮忙吗？或者有谁知道如何使用WebSocketSubject？

Answer 1

我不知道它是否仍然相关，但我确实使用了与 https://github.com/ohjames/rxjs-websockets 的角度类似的 websocket 连接

我在调用 ServerSocketService 的组件中进行了一些修改（用于连接到基于 websocket 的端点），即重试机制，并且我使用了 ReplaySubject 而不是示例中给出的 QueuingSubject。

@Injectable()
export class ServerSocket {
    // private inputStream: QueueingSubject<string>;
    private inputStream: ReplaySubject<string>;
    public messages: Observable<string>;
    private subscription: Subscription;
    private websocket: WebSocket;

    public connect() {
        if (this.messages) {
            return;
        }
        console.log('inside connect');
        // this.inputStream = new QueueingSubject<string>();
        this.inputStream = new ReplaySubject();

        // Using share() causes a single websocket to be created when the first
        // observer subscribes. This socket is shared with subsequent observers
        // and closed when the observer count falls to zero.
        this.messages = websocketConnect(
            'ws://localhost:9097/echo',
            this.inputStream
        ).messages.share();


        this.messages.retryWhen(errors => errors.delay(1000)).subscribe(message => {
           console.log('error', message);
        });
    }

    public send(message: string): void {
        // If the websocket is not connected then the QueueingSubject will ensure
        // that messages are queued and delivered when the websocket reconnects.
        // A regular Subject can be used to discard messages sent when the websocket
        // is disconnected.

        this.inputStream.next(message);

    }
}

然后在组件的 OnInit 生命周期内建立连接，订阅然后发送消息。一旦组件达到其 OnDestroy 生命周期，只需取消订阅即可释放资源。