订阅数组中所有当前和未来的项目

Question

将 RxJS 与 Angular5 结合使用，我有一个包含数组的服务。 随着数组的填充，一些组件将被创建/销毁/重新创建。

所以我们有这样的服务：

@Injectable
export class MyService {
   public events = [];
}

那么我们有一个像这样的组件：

@Inject(MyService)
@Component({})
export class MyComponent {

  mySub: Subscriber

  constructor(private ms: MyService){}

  ngOnInit(){

   this.mySub = Rx.Observable.from(this.ms.events).subscribe(v => {

   });

  }

}

我的问题是 - 如果 events 数组 aleady 中有元素，那么在创建组件时，它将获取所有现有元素，但是在创建订阅后添加到数组中的元素呢？事后如何监听元素何时添加到数组中？

如果我使用Subject，问题是我认为它不会存储历史项目，只会触发新项目。

Answer 1

我会将事件存储在您的服务中。这样，服务可以保留所有事件的历史记录，您可以使用主题来发出该历史记录。

@Injectable()
export class MyService {
    sub$ = new BehaviorSubject<any[]>([]);
    events: any[] = [];

    // add event to array and push to subject
    addEvent(event){
        this.events.push(event);
        this.sub$.next(this.events);
    }
    // find and remove event from array then push to subject
    removeEvent(event){
        let index = this.events.findIndex(item => {
            return event.id === item.id;
        });
        // if index is found, remove from array
        if(index !== -1){
            this.events.splice(index, 1);
            this.sub$.next(this.events);
        }
    }

}

我会使用行为主题，以便订阅的观察者接收最后的发射（也可以使用重放主题）。

这是一个 stackblitz 的演示

Answer 2

我认为@LLai 可以满足您的大部分需求，但我会将BehaviorSubject 更改为ReplaySubject 并将发射更改为单个事件（鉴于 Alexander Mill 的回答）。

这不包括删除事件，尽管我没有看到提到该要求。

我的服务

@Injectable()
export class MyService {
  event$ = new ReplaySubject<any>();

  addEvent(event){
    this.event$.next(event);
  }
}

我的组件

ngOnInit(){
  this.mySub = this.ms.event$.subscribe(v => {
    ...
  });
}

演示

const replaySubject$ = new Rx.ReplaySubject(); 
// Also, pass in a buffer size to stop blow-out (if applicable)
// const replaySubject$ = new Rx.ReplaySubject(maxBufferSize); 


// Events before subscribe
replaySubject$.next('event1');
replaySubject$.next('event2');

replaySubject$.subscribe(console.log);

// Events after subscribe
replaySubject$.next('event3');
replaySubject$.next('event4');

.as-console-wrapper { max-height: 100% !important; top: 0; }

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.5.6/Rx.js"></script>

Answer 3

如果你担心状态，你应该使用Subject，它可以用一些数据初始化，然后传播。

@Injectable
export class MyService {
   public events = [];
   public data$: Subject<any>;

   constructor() {
     this.data$ = this.events;
   }

   addItem(v: any) {
     this.event.push(v);
     this.data$.next(this.event);
   }
}

如果您想要实时数据源，请在您的组件中。

@Component({//usual angular stuff here})
export class MyComponent implements OnInit {
  myData: Subject<any>;

  constructor(private myService: MyService) { }

  ngOnInit() {
    this.myData = this.myService.data$;
  }
}

然后您可以使用async 管道或订阅myData 来获取您的数据。

Answer 4

好的，所以我们使用 concat 运算符合并两个 observable，它似乎可以工作。本质上，这个答案只是从头开始重新创建一个 ReplaySubject。我建议不要使用这个答案，而是使用 ReplaySubject 。无论如何：

服务看起来像这样

@Injectable
export class MyService {

   public subject = new Rx.Subject();
   public events = [];

   addItem(v: any){
    this.events.push(v);
    this.subject.next(v);
  }
}

那么组件看起来像这样

@Inject(MyService)
@Component({})
export class MyComponent {

  mySub: Subscriber

  constructor(private ms: MyService){}

  ngOnInit(){

    const obs = Rx.Observable.from(this.ms.events).concat(this.ms.subject);
    this.mySub = obs.subscribe(v => {

    });

  }

}

这个想法是将数组中的 observable 与主题连接起来，并且数组 observable 中的所有项目都会首先触发。