rxjs: observable.complete не является функцией

Question

В следующем фрагменте кода я пытаюсь продемонстрировать, как concatMap сохраняет порядок событий, даже если действие, выполняемое над событиями, не завершается по порядку.

Теперь я получаю сообщение об ошибке

delayerObservable.complete () не является функцией

Это в основном просто взято из учебника.Я вызываю next (), а затем вызываю complete ().Это должно работать, по крайней мере, так я думал.

Я могу достичь желаемой функциональности, возвращая randomDelayer.first ()

return randomDelayer.first()

, но я хотел бы завершить наблюдаемое изнутри,поскольку я могу отправить больше событий, чем одно.

const myTimer = Rx.Observable.create((observer) => {
  let counter = 0;
  setInterval( () => {
    observer.next(counter++);    
    console.log('called next with counter: '+ counter);
  },2000);
});

const myRandomDelayer = myTimer.concatMap( (value) => {
    const randomDelayer = Rx.Observable.create( (delayerObservable) => {
        const delay = Math.floor(Math.random()*1000);
        setTimeout(() => {
            console.log(delayerObservable);
            delayerObservable.next('Hello, I am Number ' + value + ' and this was my delay: ' + delay);
            delayerObservable.complete(); // <<-- this does not work (not a function)
        }, delay);
    });
    return randomDelayer;
});

myRandomDelayer.subscribe( (message) => {
    console.log(message);
});

Answer 1

Похоже, что между версиями 4 и 6 rxjs-framework'а довольно много изменений.Рабочая версия неисправного источника такова:

const { Observable } = rxjs;
const { map, filter, concatMap, pipe } = rxjs.operators;

console.log('Starting....');

const myTimer = Observable.create((observer) => {
  let counter = 0;
  setInterval( () => {
      counter++;
      if (counter < 10){
          console.log('nexting now with counter ' + counter);
          observer.next(counter);    
      } else {
          observer.complete();
      }
  },1000);
});

const myRandomDelayer = myTimer.pipe(
    concatMap( (value) => {
        const randomDelayer = Observable.create( (delayerObservable) => {
            const delay = Math.floor(Math.random()*5000);
            setTimeout(() => {
                delayerObservable.next('Hello, I am Number ' + value + ' and this was my delay: ' + delay);
                delayerObservable.complete();
            }, delay);
        });
        return randomDelayer;
    })
);

myRandomDelayer.subscribe( (message) => {
    console.log(message);
});