Stream.io Chat API-在Angular中使用带Observables/RxJS的流式聊天

Question

Stream的Chat API文档的Node版本使用async/await通过JS客户端连接到Stream Chat服务。我已经阅读了关于连接到Websockets的RxJS文档，这就是我所理解的Stream Chat客户端正在做的事情。我的问题是如何从流式聊天事件中生成Observables？例如，从事件对象(如以下对象(：

channel.on('message.deleted', event => {
console.log('event', event);
console.log('channel.state', channel.state);
});

感谢您的真知灼见。

Answer 1

如果你有一个类似套接字的连接，你可以用一个类似的Observable来包装它

function channelObs(channel) {
return new Observable<any>(
(subscriber: Observer<any>) => {
channel.on('eventId', 
event => {
subscriber.next(event);
}
);
return () => {// tearDownLogic, i.e. code executed when this Observable is unsubscribed
}
);
}

函数channelObs返回的Observable可以用RxJS 提供的所有运算符处理

Answer 2

您需要创建一个Subject并按照以下进行订阅

聊天服务.ts

private chatEventListener = new Subject<any>();
bindEvents(){
channel.on('message.deleted', event => {
const eventObj = {
event : event,
state : channel.state
};
this.chatEventListener.next(eventObj );
});
}
subscribeToNotifications(){
this.chatEventListener.asObservable();
}

在组件中。ts

this.chatService.subscribeToNotifications().subscribe(event => {
console.log(event);
})