Angular Websocket RxJS / WebSocketSubject

Question

In past we used several angular2 websocket impelmentations but we wasn't happy with them, there was several problems using them. So we decided to try our fortune doing it by our own with RxJs.

This is our first try:

@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);
    }
}

The socket is opened and working well for a while. After a few seconds the browser is closing the socket and I get an closing event on server side.

This is the closing reason we get on server site: [1006] WebSocket Read EOF

Does anyone can help? Or does anyone know how to use WebSocketSubject?

@JuliaPassynkova thanx for your link, I already found this solution and this is working great. The thing on this solution is, there is always the problem of the closing Websocket, this solutions does a reconnect on closing connection which is okay. But it is like a first connect, and on the server the first connect does a lot of expensive things. So it would be better if the connection is never closed from the browser side. So the question is, WHY the browser is closing the websocket? — Michael Burger
– Michael Burger, Commented May 30, 2017 at 12:13
Since you seems to have done your own implementation of websocket, please could you share your own solution? — BlackHoleGalaxy
– BlackHoleGalaxy, Commented Dec 26, 2017 at 14:38
Hi Michael, yes you, since you added a comment on the answer below telling you managed to solve this problem by your own implementation. We run in the same issue you were facing. — BlackHoleGalaxy
– BlackHoleGalaxy, Commented Feb 26, 2018 at 16:18

Manit · Accepted Answer · 2017-09-14 13:36:14Z

I don`t know if it is still relevant but I did use something similar for the websocket connection from angular with https://github.com/ohjames/rxjs-websockets

Some of the modifications I made inside the component where I was calling the ServerSocketService (meant for connecting to websocket based end point) i.e. Retry mechanism and I used ReplaySubject instead of the QueuingSubject given in the example.

@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);

    }
}

And then inside the component`s OnInit life cycle make the connection, subscribe and then send the messages. And once the component reaches its OnDestroy life cycle release the resources just by doing unsubscribe.

this looks good, in meanwhile we have adapt our own websocket implementation but we will have a look also on this if we have time

Collectives™ on Stack Overflow

Angular Websocket RxJS / WebSocketSubject

1 Answer 1

1 Comment

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

1 Comment

Your Answer

Sign up or log in

Post as a guest

Related