RXJS控件可观察的调用 [英] RXJS control observable invocation

问题描述

我在Angular 2项目中使用 RxJs版本5 . 我想创建一些可观察的对象，但是我不想立即调用可观察的对象.

I use RxJs version 5 within my Angular 2 project. I want to create some observables but I don't want the observables being invoked immediately.

在版本4 中，您可以使用受控命令或还 )在版本5中可用.

In version 4 you could control the invocation with (for example) the Controlled command or Pausable Buffers. But that functionality is not (yet) available in version 5.

如何在RxJs 5中获得这种功能?

How can I get the this kind of functionality in RxJs 5?

我的最终目标是将创建的可观察对象排入队列，并逐一调用它们.仅当成功处理下一个时，才调用下一个. 当一个失败时，队列将被清空.

My ultimate goal is to queue the created observables and invoke them one by one. The next one is only invoked when the previous one is processed successfully. When one fails, the queue is emptied.

编辑

使用@Niklas Fasching的评论，我可以使用

With the the comment of @Niklas Fasching I could create a working solution with the Publish operation.

JS Bin

// Queue to queue operations
const queue = [];

// Just a function to create Observers
function createObserver(id): Observer {
    return {
        next: function (x) {
            console.log('Next: ' + id + x);
        },
        error: function (err) {
            console.log('Error: ' + err);
        },
        complete: function () {
            console.log('Completed');
        }
    };
};

// Creates an async operation and add it to the queue
function createOperation(name: string): Observable {

  console.log('add ' + name);
  // Create an async operation
  var observable = Rx.Observable.create(observer => {
    // Some async operation
    setTimeout(() => 
               observer.next(' Done'), 
               500);
  });
  // Hold the operation
  var published = observable.publish();
  // Add Global subscribe
  published.subscribe(createObserver('Global'));
  // Add it to the queue
  queue.push(published);
  // Return the published so the caller could add a subscribe
  return published;
};

// Create 4 operations on hold
createOperation('SourceA').subscribe(createObserver('SourceA'));
createOperation('SourceB').subscribe(createObserver('SourceB'));
createOperation('SourceC').subscribe(createObserver('SourceC'));
createOperation('SourceD').subscribe(createObserver('SourceD'));

// Dequeue and run the first
queue.shift().connect();

推荐答案

您可以通过之后才能启动连接.

You can seperate the start of the observable from subscription to it by publishing the observable. The published observable will only be started after calling connect on it.

请注意，所有订阅者将共享可观察序列的单个订阅.

Note that all subscribers will share a single subscription to the observable sequence.

var published = Observable.of(42).publish();
// subscription does not start the observable sequence
published.subscribe(value => console.log('received: ', value));
// connect starts the sequence; subscribers will now receive values
published.connect();

这篇关于RXJS控件可观察的调用的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！