如何从所有消息的可观察对象中创建未读消息的可观察对象

Question

有一个websocket echo服务，可以发送和接收消息：

@Injectable()
export class NotificationsService {
  private ws = new $WebSocket('wss://echo.websocket.org');

  wsSendMessage(message: string = `Message ${Math.random()}`): void {
    this.ws.send(message);
  }

  wsGetMessages(): Observable<string> {
    return Observable.from(this.ws.getDataStream())
      .map((res: any) => res.data);
  }
}

Notifications 组件订阅此 observable 并在通知下拉面板中显示所有消息：

  getMessages(): void {
    this.notificationsService.wsGetMessages().subscribe(
        (message: string) => {
          this.messages.unshift(message);
        },
        (err: any) => console.log(err)
    );    
  }

当通知下拉面板打开时，我需要在导航栏组件中显示未读消息的数量并将其设置为 0。而且我不知道该怎么做，所以我需要你的帮助......

附言

现在 websocket echo 服务是：

@Injectable()
export class NotificationsService {
  private ws = new $WebSocket('wss://echo.websocket.org');

  wsSendMessage(message: string = `Message ${Math.random()}`): void {
    this.ws.send(message);
  }

  wsGetMessages(): Observable<Object> {
    return Observable.from(this.ws.getDataStream())
      .map((res: any) => res.data);
  }

  messagesReadStream(msgArray: Object[]): Observable<Object> {
    return Observable.from(msgArray);
  }
}

通知组件是：

export class Notifications implements OnInit {
  messagesAll:string[] = [];
  messagesRead:string[] = [];
  newMessagesNumber:number = 0;

  constructor(private notificationsService: NotificationsService) {
  }

  addMessage(): void {
    this.notificationsService.wsSendMessage();
  }

  getMessages(): void {
    this.notificationsService.wsGetMessages()
      .subscribe(
        (message: any) => this.messagesAll.unshift(message),
        (err: any) => console.error(err)
      );
  }

  readMessages(): void {
    this.messagesRead = this.messagesAll.slice();

    this.notificationsService.messagesReadStream(this.messagesRead)
      .switchMap(() => 
        this.notificationsService
          .wsGetMessages()
          .scan((unread, _) => unread + 1, 0)    
          .startWith(0))
      .subscribe(
        (count: number) => this.newMessagesNumber = count,
        (err: any) => console.error(err)
      );    
  }

  ngOnInit(): void {
    this.getMessages();
  }  
}

带模板：

<button type="button" class="btn btn-primary-outline btn-sm" (click) = "addMessage()">
  Add message
</button>  

<button type="button" class="btn btn-primary-outline btn-sm" (click) = "readMessages()">
  Read messages
</button>  

<div style="margin-top: 1rem">
  <span class="circle bg-warning fw-bold">
      {{newMessagesNumber}}
  </span>  
</div>

现在，如果我单击 button 'Add message'，我的计数器不会从 0 更改值。只有当我单击button 'Add message'，然后单击button 'Read messages'，然后再次单击button 'Add message'，计数器才开始正常工作。你能解释一下，问题是什么吗？

Answer 1

您将需要多个Observable。一个用于包含计数，一个用于在读取消息时发出信号。前者将使用wsGetMessage 流，而后者必须连接到一个 Observable，该 Observable 会在通知面板打开时触发。

const panelOpened: Observable<any> = //..Need a signalling mechanism for when the messages get read

const unreadCount: Observable<number> = messagesRead.switchMap(() => 
   this.notificationsService
         .wsGetMessages()
         //Just count all of the messages as they come in.
         .scan((unread, _) => unread + 1, 0)
         //Make sure it always gets reset to zero
         .startWith(0));

//Subscribe to updates on count changes.
unreadCount.subscribe(count => this.unreadMessageCount = count);