RxJS配置可观察到的多播

Question

我正在努力理解rxJs观察者/可观察的。

我有一个Angular应用程序，其中一个Inject table服务创建一个可观察的：

闲聊服务ts:

public chatObserver: Observer<any>;
public chatObservable: Observable<any>;
constructor() {
  this.chatObservable = Observable.create((observer: Observer<any>) => {
  this.chatObserver = observer;
  });
}

当聊天室中收到消息时。服务ts，消息（假定）是多播的：

this.chatObserver.next(chatMessage);

两个组件以这种方式订阅可观察对象：

constructor(private _chatService: ChatService) {}

ngOnInit() {
    this._chatService.chatObservable.subscribe(
      (message) => {
        console.log('New chat message received!', message)
        // Do something with message
      }
    );

问题：只有一个（最后一个订阅的）收到消息。

问：我如何配置可观察到多播到所有订阅者？

Answer 1

一个简单的复习。如果您想在订阅时（在服务/管道等中手动，或通过html模板中的异步管道）接收一些初始值/当前值，请使用行为主题。这里有一个关于你试图实现的目标的基本建议。当然，考虑到生产，还有很多事情要做。问候。

闲聊服务ts:

BackEndDB : ChatMessage[] = <Obtain data for chatHistory from your DB>();

如果您将任何后端存储用于聊天历史记录，请通过该源初始化BackEndDB，如果不是则将消息$设置为新行为主体

messages$ = new BehaviorSubject<ChatMessage[]> // (this.BackEndDB OR null);
sendMessage(message: ChatMessage): void {
  this.BackEndDB.push(message);
  this.messages$.next(this.BackEndDB);
}
getMessages(): Observable<ChatMessage[]> {
  return this.messages$.asObservable();
}

然后在组件中：

import {ChatService} from 'yourPath/chat.service'; 

@Component({ 
  selector: 'app-chat',
  templateUrl: ['./chat.component.html'],
  styleUrls: ['./chat.component.css'],
})
export class ChatComponent {
  inbox$: Observable<ChatMessage[]>;
  constructor(private chatService: ChatService) {
    this.inbox$ = this.chatService.getMessages(); 
  } 
  sendMessage(username, msg) {
    let message: ChatMessage = { 
      msg: msg, 
      sender: username 
    }
    this.chatService.sendMessage(message);
  }
} 

然后聊天。组成部分html:

<label> Your Name:  </label>
<input type="text" #userName>
<label> Your Message:  </label>
<input type="text" #userMessage>  
<button type="button" (click) = "sendMessage(userName.value,userMessage.value)">Send</button>
<div *ngIf="inbox$ | async; let messages ">
  <div *ngFor= "let message of messages ">
    <p> {{ message.sender  }} </p>
    <p> said: </p>
    <p> {{ message.msg }} </p>
  </div>
</div>

如果要重用ChatMessage接口，可以在共享文件夹中创建该接口，如果该接口仅在共享文件夹中使用，则可以直接在组件中创建：

interface ChatMessage {
  msg: string;
  sender: string;
}

我们将新消息推送到BackEndDB变量并调用Next的唯一原因是因为这是一个简单的演示。在现实中，您将使用FirestoreCollection Document.value更改（）或任何其他实时数据服务。

Answer 2

下面是一个如何使用多播观察者的示例：

//chat.service.ts：

public chatObserver: Subject<any>;
constructor() {
  this.chatObserver = new Subject();
}

发出聊天信息（受试者设计了next、error和complete方法）：

this.chatObserver.next(chatMessage);

在组件中订阅：

constructor(private _chatService: ChatService) {}

ngOnInit() {

  this._chatService.chatObserver.subscribe(
    (message) => {
      console.log('New chat message received!', message)
      // Do something with message
    }
  );

注意：但是当一些组件在发出聊天消息后订阅聊天观察者时，我建议使用行为主体而不是行为主体（只需将主体替换为行为主体）。

BehaviorSubject具有存储“当前”值的特性。这意味着您始终可以直接从BehaviorSubject（源/引号：https://medium.com/@luukgruijs/understanding-rxjs-behaviorsubject-replaysubject-and-asyncsubject-8cc061f1cfc0）