rxjs5 / Angular-清除ReplaySubject缓冲区

Question

我有一个Angular服务，它异步地将数据流（http调用）共享到几个组件。我需要偶尔基于用户操作来重新调用http服务。

我正在使用ReplaySubject保存加载的值并发送给在http调用后进行订阅的订阅者。

我想知道是否可以在进行后续的http调用之前清除ReplaySubject的缓冲区？在此期间，我怀疑是否还需要取消订阅才能不产生泄漏？

服务：

@Injectable()
export class GreatDataService {

    public data$: ReplaySubject<any>;
    private subs: Subscription;

    constructor(private http: Http) {
        this.data$ = new ReplaySubject(1);
    }

    public refresh() {
        if (this.subs) {
            this.subs.unsubscribe();
            this.subs = null;
        }
        this.subs = this.http.get('/api').subscribe(this.data$)
    }
}

顶级部分组件：

...
    constructor(private greatDataService: GreatDataService) {}
    ngOnInit() {
         this.greatDataService.refresh();
    }
...

组件1：

...
    constructor(private greatDataService: GreatDataService) {}
    ngOnInit() {
    this.greatDataService.data$.subscribe(
        x => console.log('subscriber 1: ' + x),
        err => console.log('subscriber 1: ' + err),
        () => console.log('subscriber 1: Completed')
    );
...

组件2：

...
    constructor(private greatDataService: GreatDataService) {}
    ngOnInit() {
    this.greatDataService.data$.subscribe(
        x => console.log('subscriber 2: ' + x),
        err => console.log('subscriber 2: ' + err),
        () => console.log('subscriber 2: Completed')
    );
...

Answer 1

您可以使用 Rx.Subject 来发出新的“get-fresh-data”事件，以便您可以在refresh()使用 .Subject 调用时检索新数据.switchMap()。请参阅此示例如何执行此操作：

function getData() {
  return Rx.Observable.of('retrieving new data')
    .timestamp()
    .delay(500);
}

// in this example i use an eventStream of clicks
// you can use Rx.Subject() and manually .next() a new value
// when somebody invokes .refresh()
const refreshDataClickStream = Rx.Observable.fromEvent(document.getElementById('refresh_stream'), 'click');

const dataStream = refreshDataClickStream
  .startWith('PAGE_LOAD') /* let the stream always first time fetching data */
  .switchMap(() => getData()) /* getData() is not cached so we switchMap to a new instance, abandoning the previous result*/
  .publishReplay().refCount(); /* refCounter so everybody gets the same results */

dataStream.subscribe(data => console.log('sub1 data: ' + JSON.stringify(data)));

setTimeout(() => {
  console.log('late arriving subscription (gets same stream)');
  dataStream.subscribe(data => console.log('sub2 data: ' + JSON.stringify(data)));
}, 2000);

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script>
<input type='button' id='refresh_stream' value="refresh_stream" />