Переход с RxJS4 на RxJS5 — реализация Observer

Из источника:

export interface Observer<T> {
  closed?: boolean;
  next: (value: T) => void;
  error: (err: any) => void;
  complete: () => void;
}

Observer<T> — это интерфейс с методами onNext, onCompleted и onError. Интерфейс — это всего лишь языковая конструкция. Он просто используется компилятором машинописного текста для проверки типов объектов, требующих Observer<T>. Он стирается при компиляции.

Класс Subscriber<T> реализует интерфейс Observer<T>. Это означает, что Subscriber<T> является фактическим конкретным классом с указанными выше методами.

Поэтому вместо этого вы используете var push = Rx.Subscriber.create(v => { [...].

Примечание:

В исходной реализации Rx интерфейсы были IObservable<T> и IObserver<T> и использовались методы расширения для обеспечения возможности композиции. Когда дело дошло до JS, у них должны были быть методы прототипа самого Observable/Observer, чтобы включить композицию, поэтому у самого класса были методы.

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

Он имеет статический метод Subscriber.create. См. https://github.com/ReactiveX/rxjs/blob/master/src/Subscriber.ts#L32

Этот метод возвращает объект Subscriber, который впоследствии можно использовать, например obsEnqueue.subscribe(push);.

Почему бы просто не встроить функцию onNext прямо в подписку?

this.queueStream = Rx.Observable.create(obs => {
    return obsEnqueue.subscribe(
      v => {
        if ((index % obsEnqueue.observers.length) === obsEnqueue.observers.indexOf(push)) {
            obs.next(v);
        }
      }
    );
});

Если бы я понял, что этот частичный код пытается сделать...
Я думаю, что нет,
я не вижу, как можно сделать это "упрощением".
возможно, что здесь нужно улучшить
заключается в том, чтобы сделать его более "повторно используемым",
сделать его модулем ?,
или, возможно, оператором Rx, если его еще нет...

это может быть попытка

/*
    "dependencies": {
        "rxjs": "^5.0.2"
    }
*/

import {Observable, Observer, Subject, Subscriber} from "rxjs";

export interface ICircularQueue<T> extends Observable<T> {
    push(value: T): void;
}

/**
* on every push use 'NEXT' observer/subscription,
* in the order they've been subscribed,
* cycling back to 1st subscription after last
*/
export function create<T>(): ICircularQueue<T> {

    let index = 0;

    let obsEnqueue = new Subject<T>();

    let queueStream = Observable.create((obs: Observer<T>) => {

        let push = Subscriber.create<T>(v => {
            // ! error ?
            if ((index % obsEnqueue.observers.length) === obsEnqueue.observers.indexOf(push)) {
                obs.next(v);
            }
        });

        return obsEnqueue.subscribe(push);
    });

    queueStream.push = (v: T) => {
        obsEnqueue.next(v);
        index++;
    };

    return queueStream;
}

потом базовый тест....

import * as CircularQueue from "./CircularQueue";
import * as assert from "assert";

const $in = (array: any[], x: any) => {
    for (let $x of array) {
        if ($x === x) { return true; }
    }
    return false;
};

describe("CircularQueue", () => {

    it("works", () => {
        let queue = CircularQueue.create();

        let result: number[] = [];
        queue.subscribe(x => {
            assert.ok($in([0, 4, 8], x));
            result.push(0);
        });
        queue.subscribe(x => {
            assert.ok($in([1, 5, 9], x));
            result.push(1);
        });
        queue.subscribe(x => {
            assert.ok($in([2, 6, 10], x));
            result.push(2);
        });
        queue.subscribe(x => {
            assert.ok($in([3, 7, 11], x));
            result.push(3);
        });

        for (let i = 0; i < 12; i++) {
            queue.push(i);
        }

        assert.equal(result.join(), "0,1,2,3,0,1,2,3,0,1,2,3");
    });
});