rxjs5 / Angular – 清除ReplaySubject缓冲区

发布时间：2020-12-17 17:00:49 所属栏目：安全来源：网络整理

导读：我有一个Angular服务,它以异步方式将数据流(http调用)共享给多个组件.我需要偶尔根据用户操作回忆一下http服务. 我正在使用ReplaySubject来保存加载的值并发送给在http调用之后订阅的订阅者. 我想知道在进行后续的http调用之前是否有办法清除ReplaySubject的

我有一个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')
    );
...

解决方法

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

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" />

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!