Angular 2 sharing web services through components

Question

I am building a web application using angular 2 in which I want multiple components to listen on the same service. This service returns an observable that returns incoming data from the web layout. I wrote code based on this example.

Current problem: Data is sent from the home component through the service to the server (using web maps), and the data is returned. However, only the observer in home.component receives the call (with id: room.created and data), and not the one that is in the navigation bar.

Can someone tell me why not both are being called? I also tried adding $ .subscribe messages to app.component, but to no avail.

Now let's move on to the code.

A message service that returns an observable. This service is used by components to send and receive messages.

@Injectable()
export class MessageService {
    private _messages: Rx.Subject<Message>;
    messages$: Rx.Observable<Message>;

    constructor(wsService: SocketService, private configuration: Configuration) {
      console.log('messag eservice');
      this._messages = <Rx.Subject<Message>>wsService
        .connect()
        .map((response: MessageEvent): Message => {
            let data = JSON.parse(response.data);
            return {
                id: data.id,
                data: data.data,
            }
        });

      this.messages$ = this._messages.asObservable();
    }

    public send(message: Message): void {
      this._messages.next(message);
    }
} 

A socket service that creates a connection to a websocket and associates itself with the input and output of this socket.

import { Injectable } from '@angular/core';
import * as Rx from "rxjs/Rx";
import { Configuration } from '../app.constants';

@Injectable()
export class SocketService {
    private subject: Rx.Subject<MessageEvent>;

    constructor(private configuration: Configuration){};

    public connect(wsNamespace = ''): Rx.Subject<MessageEvent> {
        var url = this.configuration.wsUrl + wsNamespace;
        if(!this.subject) {
            this.subject = this.create(url);
        }
        return this.subject;
    }

    private create(url): Rx.Subject<MessageEvent> {
        let ws = new WebSocket(url);

        // bind ws events to observable (streams)
        let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => {
            ws.onmessage = obs.next.bind(obs);
            ws.onerror = obs.error.bind(obs);
            ws.onclose = obs.complete.bind(obs);

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

        // on obs next (send something in the stream) send it using ws.
        let observer = {
            next: (data: Object) => {
                if (ws.readyState === WebSocket.OPEN) {
                    ws.send(JSON.stringify(data));
                }
            },
        };

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

Application component with the following providers:

  providers: [MessageService, SocketService, Configuration, AuthService]

I am instantiating providers in my main app.component application to make sure messages and socket services are not created twice.

My home.component looks like this (this is a page loaded using routing):

import { Component, OnInit } from '@angular/core';
import { Subject } from 'rxjs';
import { Router }    from '@angular/router';
import { MessageService } from '../../services/message.service';

@Component({
  ...
  providers: []
})
export class HomeComponent implements OnInit {
  constructor(private router: Router, private messageService: MessageService) {}

  ngOnInit(): void {
    this.messageService.send({
      id: 'room.create',
      data: {'name': 'Blaat'}
    });

    this.messageService.messages$.subscribe(msg => {
      console.log(msg);
        if(msg.id == 'room.created') {
            // navigate naar games!
        }
    });
  }

}

My navigator component looks like this (directive):

import { Component, OnInit } from '@angular/core';
import { MessageService } from '../../services/message.service';

@Component({
  moduleId: module.id,
  selector: 'navbar',
  templateUrl: 'navbar.component.html',
  styleUrls: ['navbar.component.css']
})
export class Navbar implements OnInit {

  constructor(private messageService: MessageService) { }

  ngOnInit() {

    this.messageService.messages$.subscribe(msg => {
      console.log(msg);
        if(msg.id == 'room.created') {
            // navigate naar games!
        }
    });
  }

}

Answer 1

, create , , = > = > . , fn websocket onmessage, onerror onclose. , ( ).

        // bind ws events to observable (streams)
        let observable = Rx.Observable.create((obs: Rx.Observer<MessageEvent>) => {
            ws.onmessage = obs.next.bind(obs);
            ws.onerror = obs.error.bind(obs);
            ws.onclose = obs.complete.bind(obs);

            return ws.close.bind(ws);
        }).share();

, https://github.com/ReactiveX/rxjs/blob/master/src/observable/dom/WebSocketSubject.ts https://github.com/blesh/RxSocketSubject