Что случилось с Observable.transduce в RxJS v5 +?

У меня был точно такой же вопрос, и я нигде не мог найти ответа. Да, вы можете pipe, но я считаю, что это создаст промежуточные наблюдаемые для каждого оператора. Хотя я не знаю наверняка, речь идет о чтении кода.

Так что я придумал свой собственный transduce оператор:

function transformForObserver(o) {
  return {
    "@@transducer/init": function() {
      return o;
    },
    "@@transducer/step": function(obs, input) {
      return obs.next(input);
    },
    "@@transducer/result": function(obs) {
      return obs.complete();
    }
  };
}

    const transduce = (obs, transducer) => {
      const xform = transducer(transformForObserver);

      return Observable.create(o => {
        return obs.subscribe({
          next: x => {
            const res = tryCatch(
              xform["@@transducer/step"],
              err => {
                console.error(`Error occurred in transducer/step!`, err);
                return err;
              }
            )(xform, o, x);

            if (res instanceof Error) { o.error(res); }
          },
          error: err => {
            console.error(`Error occurred in observable passed to Rx transduce fn!`, err);
            o.error(err);
          },
          complete: () => {o.complete();}
        });
      });
    }

Еще не тестировал, скоро напишу об этом, если будет интерес.

Обновление: я разветвил библиотеку tranducers jslongser и включил в нее такие преобразователи. Форк - это https://github.com/brucou/transducers.js, а функция transduceLazyObservable. Ср. тесты для примера использования.