angular - 使用 WebSocket 可观察到的 RxJs

Question

我的 Angular 应用程序使用 websocket 与后端通信。

在我的测试用例中，我有 2 个客户端组件。 Observable 计时器按预期打印两个不同的客户端 ID。

每个 ngOnInit() 还打印其客户端的 ID。

现在由于某种原因，websocketService.observeClient() 的订阅会为每条消息调用 2 次，但 this.client.id 始终打印第二个客户端的值。

这是我的客户端组件

@Component({
...
})
export class ClientComponent implements OnInit {

  @Input() client: Client;

  constructor(public websocketService: WebsocketService) {
    Observable.timer(1000, 1000).subscribe(() => console.log(this.client.id));
  }

  ngOnInit() {

    console.log(this.client.id);
    this.websocketService.observeClient().subscribe(data => {
      console.log('message', this.client.id);
    });

  }

}

还有我的 websocket 服务

@Injectable()
export class WebsocketService {

  private observable: Observable<MessageEvent>;
  private observer: Subject<Message>;

  constructor() {

    const socket = new WebSocket('ws://localhost:9091');

    this.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);
      }
    );

    this.observer = Subject.create({
      next: (data: Message) => {
        if (socket.readyState === WebSocket.OPEN) {
          socket.send(JSON.stringify(data));
        }
      }
    });

  }

  observeClient(): Observable<MessageEvent> {
    return this.observable;
  }

}

---

编辑

好的，据我所知，它与 Observables 是单播对象这一事实有关，我必须为此使用 Subject，但我不知道如何创建 Subject。

Answer 1

从 rxjs 5 开始，您可以使用内置的 websocket 功能为您创建主题。当您在出错后重新订阅流时，它也会重新连接。请引用这个答案:

https://stackoverflow.com/a/44067972/552203

简单描述:

 let subject = Observable.webSocket('ws://localhost:8081');
 subject
   .retry()
   .subscribe(
      (msg) => console.log('message received: ' + msg),
      (err) => console.log(err),
      () => console.log('complete')
    );
 subject.next(JSON.stringify({ op: 'hello' }));