将 Web 套接字消息过滤到多个 Observable 中答案

【问题标题】：Filter web socket messages into multiple Observables将 Web 套接字消息过滤到多个 Observable 中
【发布时间】：2019-07-12 17:13:25
【问题描述】：

在我的 Angular 应用程序中，我使用 Web 套接字从后端接收异步消息。所以我创建了一个角度服务来处理这个问题：

import { Injectable } from '@angular/core';
import { Observable } from 'rxjs';
import * as SocketIO from 'socket.io-client';

@Injectable({
    providedIn: 'root'
})

export class EventService {
    private socket;

    constructor() {
        this.socket = SocketIO();
    }

    onMessage(): Observable<any> {
        return new Observable<any>(observer => {
            this.socket.on('message', (message: string) => {
                observer.next(JSON.parse(message))
            });
        });
    }
}

我的消息有不同的类型，并且采用这种 JSON 格式：

{
    "type": "HELLO",
    "data": "hello world !"
}

目前，我使用EventService.onMessage().subscribe(...) 函数来处理唯一需要它们的组件中的消息。

现在，我需要有多种消息类型。这将被多个组件使用。

我想创建一个按消息类型返回 Observable 的方法。类似的东西：

export class EventService {
    ...

    onHello(): Observable<any> { ... }

    onFoo(): Observable<any> { ... }

    onBar(): Observable<any> { ... }
}

有可能吗？怎么做？后端架构是否正确（也许我应该拆分成多个 Web 套接字）？

【问题讨论】：

我可能无法使我的标题相关。随意编辑它。

标签： angular rxjs observable

【解决方案1】：

也许可以尝试不同的方式：为每种消息类型创建主题，并在 onMessage 方法中将消息应用于特定主题

onHello() = new Subject()/new BehaviorSubject(initialValue)
...

onMessage() {
  this.socket.on('message', message => {
    let msg = JSON.parse(message);
    if(msg.type === 'hello'){
      this.onHello.next(msg);
    }
    ...
  }
}

【讨论】：

我找到了一个更简单的解决方案，但感谢您指出RxJS Subject。

【解决方案2】：

事实上，当我使用socket.io 时，有一个简单的解决方案。

我正在使用send() 函数在我的后端发送数据：

const io = require('socket.io');
const ws = io(...);

ws.on('connection', function(socket) {
    ...

    socket.send(JSON.stringify({
        type: "HELLO",
        data: "hello world !"
    }));
});

发送'message' 可以使用on('message', ...) 接收。

但我应该改用emit() 函数来设置事件名称：

const io = require('socket.io');
const ws = io(...);

ws.on('connection', function(socket) {
    ...

    socket.emit('hello', JSON.stringify({
        data: "hello world !"
    }));
});

所以现在我的角度服务可能是：

export class EventService {
   ...

    onHello(): Observable<any> {
        return new Observable<any>(observer => {
            this.socket.on('hello', (message: string) => {
                observer.next(JSON.parse(message))
            });
        });
    }
}

【讨论】：