RXJS 中的分页数据光标以及对 subject.onDone 和错误的混淆

我正在使用RXJS，并提出了分页数据光标的实现。我没有花太多时间在响应式函数式编程上，我想知道我的实现是否符合这个库的使用方式。

我想要一个可以从端点加载页面的类。如果您订阅它，您将收到查询的最后一页。第一个订阅会导致自动查询第一页。对"getPage"的调用应为所有订阅触发onNext。多个订阅不应导致多个请求。

我写了一个满足这一点的基本例子，在我的思考过程中进行了大量评论：https://jsfiddle.net/gfmn708g/1/

我的问题是：

• 这符合RXJS的精神吗？同时使用 ReplySubject 和 shareReplay 对我来说感觉不对，但我找不到其他方法来获得我想要的行为。我读到使用主题是"不好的"，违背了范式的原则。
完成
• 并处理所有正在进行的请求后，第 63 行是否会取消订阅/完成所有 items$ 订阅(第 82 行和第 89 行(？
处理错误的正确
• 方法是什么，以便将错误传播到订阅者，但它们不会谋杀流并阻止我推送更多请求？

(这是根据SO的问题指南列出的代码(

const logDiv = $("#log");
function log(message, cls) {
    logDiv.append($("<li>").text(message).addClass(cls));
}
/* interface IRequest {
    url: string;
    page: number:
    refresh?: boolean
}
interface IEndpoint {
    get(request: IRequest): [];
} */
// Class that represents a cursor into paginated data
function PagedData(endpoint, url) {
    this._endpoint = endpoint;
    this._url = url;
    // Our request queue is an observable of structurs of type IRequest
    // We use a reply subject so that the last URL requested is in the stream when the first subscriber subscribes.
    this._requestQueue = new Rx.ReplaySubject(1);
    // This is our data observable, subscribe to it to
    // A) receive the last page that this cursor has produced
    // B) receive all future pages
    this.items$ = this._requestQueue
        // Don't re-query unless the "refresh" boolean is true
        .distinctUntilChanged(req => req, (left, right) => right.refresh ? false : left.page == right.page)
        // Make the request...
        .flatMapLatest(request => Rx.Observable.of(request).zip(this._endpoint.get(request)))
        // Wrap data returned with an envelope with data such as which page was requested
        .map(data => {
            const request = data[0];
            const response = data[1];
            return {
                page: request.page,
                url: request.url,
                items: response
            };              
        })
        // Replay last page worth of data on each subscription
        .shareReplay(1);
    // Queue up the first page to be retrieved on first subscription
    this.getPage(1);
}
PagedData.prototype.getPage = function(page, refresh) {
    refresh = refresh || false;
    // Fire off the workflow
    this._requestQueue.onNext({
        url: this._url,
        refresh: refresh,
        page: page
    });
}
PagedData.prototype.dispose = function() {
    // Question: this should unsubscribe ALL of the subscriptions to this.items$, right?
    this._requestQueue.completed();
}
// -----------------
// EXAMPLE USAGE
var dummyEndpoint = {
    get(request) {
        log(`GET: ${request.url} at page ${request.page}`, "service");
        return Rx.Observable.range(request.page * 10, 10)
            .delay(1000)
            .map(i => ({id: i, title: `Track ${i}`}))
            .toArray();
    }
};
const tracks = new PagedData(dummyEndpoint, "/api/tracks");
// This results in getting the first page
tracks.items$.subscribe(data => {
    log(`On page ${data.page}, ${data.items.map(i => i.title).join(",")}`, "first")
});
// Wait one second after getting the first page
window.setTimeout(() => {
    // Subscribe again, we will receive the first page with no re-query
    tracks.items$.subscribe(data => log(`Got page ${data.page} after delay`, "second"));
    // Get the second page
    tracks.getPage(2);
    // Wait another second after getting the second page
    window.setTimeout(() => {
        log("Getting second page (without refresh)");
        // This shouldn't result in anything, since "refresh" is false/undefined
        tracks.getPage(2);
        // Wait one more second...
        window.setTimeout(() => {
            log("Getting second page (with refresh)");
            // This should result in getting the second page, refresh is true
            tracks.getPage(2, true);
      // Should get rid of all subscriptions after the last in-flight request?
            tracks.dispose();
        }, 1000);
    }, 2000);
}, 2000);