rxjs ReplaySubject 句柄

Question

我在 angular 4 中使用的模板有问题。此模板实现了一个通知系统，您可以在其中添加新的通知，但文档未指定如何删除观察者 ReplaySubject 的元素。

模板将此作为服务实现如下：

private notificationsList: Notification[] = [];
  // a stream that publishes new notifications only once
  public newNotifications: Subject<Notification> = new Subject<Notification>();

  // `notifications` is a stream that emits an array of the most up to date notifications
  public notifications: ReplaySubject<Notification[]> =
      new ReplaySubject<Notification[]>(1);

  // `updates` receives _operations_ to be applied to our `notifications`
  // it's a way we can perform changes on *all* notifications (that are currently
  // stored in `notifications`)
  public updates: Subject<any> = new Subject<any>();

  // action streams
  public create: Subject<Notification> = new Subject<Notification>();
  // public markThreadAsRead: Subject<any> = new Subject<any>();

  constructor() {
    // recois des operation, et les fais sur la liste interne, puis diffuse le
    // resultat sur notifications
    this.updates.subscribe((ope) => {
      this.notificationsList = ope(this.notificationsList);
      console.log(this.notificationsList);
      this.notifications.next(this.notificationsList);
    });

    this.newNotifications
      .map(function(notification: Notification): INotificationsOperation {
        return (notifications: Notification[]) => {
          return notifications.concat(notification);
        };
      })
      .subscribe(this.updates);

  }

  // an imperative function call to this action stream
  public addNotification(notification: Notification): void {
    this.newNotifications.next(notification);
  }

我试图询问所有者如何删除通知列表的实际元素，但他只是告诉我我可以访问 "notifications" 主题以接收它的最新版本。但不要提及我如何实际删除列表中的一个元素。

有人知道吗？

谢谢！

Answer 1

我添加了一个public功能，你可以使用。例如，如果您想按名称删除元素，或者不想调整列表大小，我添加了一条注释，让您了解可以修改代码的哪一部分。我的 post.

末尾的解释

  private notificationsList: Notification[] = [];
  // a stream that publishes new notifications only once
  public newNotifications: Subject<Notification> = new Subject<Notification>();
  public removeNotificationByIndex$ : Subject<number> = new Subject<number>();
  // `notifications` is a stream that emits an array of the most up to date notifications
  public notifications: ReplaySubject<Notification[]> =
      new ReplaySubject<Notification[]>(1);

  // `updates` receives _operations_ to be applied to our `notifications`
  // it's a way we can perform changes on *all* notifications (that are currently
  // stored in `notifications`)
  public updates: Subject<any> = new Subject<any>();

  // action streams
  public create: Subject<Notification> = new Subject<Notification>();
  // public markThreadAsRead: Subject<any> = new Subject<any>();

  constructor() {
    // recois des operation, et les fais sur la liste interne, puis diffuse le
    // resultat sur notifications
    this.updates.subscribe((ope) => {
      this.notificationsList = ope(this.notificationsList);
      console.log(this.notificationsList);
      this.notifications.next(this.notificationsList);
    });

    this.newNotifications
      .map(function(notification: Notification): INotificationsOperation {
        return (notifications: Notification[]) => {
          return notifications.concat(notification);
        };
      })
      .subscribe(this.updates);

    this.removeNotificationByIndex$
     .map(function(index: number){
        return (notifications: Notification[]) => {
        // >>>> DELETE METHOD IS TO BE DEFINED DOWN HERE !
        notifications.splice(index,1);
        // >>>> DELETE METHOD IS TO BE DEFINED UP HERE !
      return notifications
     };
     })
     .subscribe(this.updates);

  }

  // an imperative function call to this action stream
  public addNotification(notification: Notification): void {
    this.newNotifications.next(notification);
  }

  // delete the element in the "index" position of the list. 
  // /!\ Resizes the list 
  public removeNotificationsByIndex(index: number): void {
    this.removeNotificationByIndex$.next(index);
  }

有什么变化？

public removeNotificationByIndex$ : Subject<number> = new Subject<number>();

该主题将（异步）接收一个索引，并使用该索引触发一个进程。

 this.removeNotificationByIndex$
 .map(function(index: number){
    return (notifications: Notification[]) => {
    // >>>> DELETE METHOD IS TO BE DEFINED DOWN HERE !
    notifications.splice(index,1);
    // >>>> DELETE METHOD IS TO BE DEFINED UP HERE !
  return notifications
 };
 })
 .subscribe(this.updates);

当索引被发出时（即你使用相关的命令式函数），一个函数（ES6 箭头函数）从中生成。就是这样：

(notifications: Notification[]) => {
    // >>>> DELETE METHOD IS TO BE DEFINED DOWN HERE !
    notifications.splice(index,1);
    // >>>> DELETE METHOD IS TO BE DEFINED UP HERE !
  return notifications
 };

这个函数被传递给this.update，它将被应用。在这个上下文中，ope 就是这个函数。收到后，this.notificationList 修改如下：

this.notificationsList = ope(this.notificationsList);

最后，这个新列表被发布到 ReplaySubject notifications:

this.notifications.next(this.notificationsList);

将这个新列表传播给它的所有订阅者。

瞧瞧 :)。祝你好运！

rxjs ReplaySubject 句柄

rxjs ReplaySubject handle

javascript

subject

rxjs

angular