如何在 TypeScript 中创建扩展 Rx.Observable 的 class？

Question

我已经使用了来自 Definitely Typed 的 rx.js 打字的最新版本。

当我尝试这个时：

class MyObservable extends Rx.Observable<any> { }

我得到了：A class may only extend another class.

为什么 Observable 和 Subject 等定义为 rx.d.ts 中 class 的接口实例？

如果我想创建一个扩展 Observable 或 Subject 的 class，我该怎么做？

P.S。我想让这个 class 处理特定领域的逻辑，所以我需要创建一个新的 class，而不是直接更新 Observable 的原型。

谢谢！

Answer 1

rx.js 中的底层 Observable 无法扩展，因为它更像是 TypeScript module 而不是 class（即它是一个单例）。

 var Observable = Rx.Observable = (function () {
     //...
 })();

这就是为什么在 Definitely Typed 定义中将其建模为接口而不是 class 的原因。要实现接口，您必须提供与接口兼容的结构。这是 IObservable<T>.

的示例

class MyObservable<T> implements Rx.IObservable<T> {
    subscribe(observer: Rx.Observer<T>): Rx.IDisposable;
    subscribe(onNext?: (value: T) => void, onError?: (exception: any) => void, onCompleted?: () => void): Rx.IDisposable;
    subscribe(a?: Rx.IObserver<T> | Function, onError?: (exception: any) => void, onCompleted?: () => void) {
        return null;
    }

    subscribeOnNext(onNext: (value: T) => void, thisArg?: any): Rx.IDisposable {
        return null;
    }

    subscribeOnError(onError: (exception: any) => void, thisArg?: any): Rx.IDisposable {
        return null;
    }

    subscribeOnCompleted(onCompleted: () => void, thisArg?: any): Rx.IDisposable {
        return null;
    }
}

Answer 2

我不得不为 WebRx 解决同样的问题。正如您已经发现的那样，使用 Typescript class 扩展 RxJS 的 IObservable 不是一种选择，因为 Observable 正在作为接口导出。正如我在对 Steve Fenton 的回答的评论中提到的，创建一个实现 Rx.IObservable 的 class 也不会让你走得太远，因为绝大多数 Rx 运算符都是围绕 [=27= 定义的] 派生自 Rx.IObservable 的接口。您几乎会重写 Rx.Observable.

在找到更好的方法之前，我解决问题的方法是使用 prototypal inheritance and export the extension via a custom d.ts 文件扩展内置 Rx.Observable：

RxExtension.ts

var RxObsConstructor = (<any> Rx.Observable);   // this hack is neccessary because the .d.ts for RxJs declares Observable as an interface)

/**
* Creates an read-only observable property with an optional default value from the current (this) observable
* (Note: This is the equivalent to Knockout's ko.computed)
* @param {T} initialValue? Optional initial value, valid until the observable produces a value
*/
RxObsConstructor.prototype.toProperty = function(initialValue?: any, scheduler?: Rx.IScheduler) {
    scheduler = scheduler || Rx.Scheduler.currentThread;

    // initialize accessor function (read-only)
    var accessor: any = (newVal?: any): any => {
        if (arguments.length > 0) {
            internal.throwError("attempt to write to a read-only observable property");
        }

        if (accessor.sub == null) {
            accessor.sub = accessor._source.connect();
        }

        return accessor.value;
    };

    //////////////////////////////////
    // IUnknown implementation

    accessor.queryInterface = (iid: string) => {
        if (iid === IID.IUnknown ||
            iid === IID.IObservableProperty ||
            iid === IID.IDisposable)
            return true;

        return false;
    };

    //////////////////////////////////
    // IDisposable implementation

    accessor.dispose = () => {
        if (accessor.sub) {
            accessor.sub.dispose();
            accessor.sub = null;
        }
    };

    //////////////////////////////////
    // IObservableProperty<T> implementation

    accessor.value = initialValue;

    // setup observables
    accessor.changedSubject = new Rx.Subject<any>();
    accessor.changed = accessor.changedSubject
        .publish()
        .refCount();

    accessor.changingSubject = new Rx.Subject<any>();
    accessor.changing = accessor.changingSubject
        .publish()
        .refCount();

    accessor.source = this;
    accessor.thrownExceptions = internal.createScheduledSubject<Error>(scheduler, app.defaultExceptionHandler);

    //////////////////////////////////
    // implementation

    var firedInitial = false;

    accessor.sub = this
        .distinctUntilChanged()
        .subscribe(x => {
            // Suppress a non-change between initialValue and the first value
            // from a Subscribe
            if (firedInitial && x === accessor.value) {
                return;
            }

            firedInitial = true;

            accessor.changingSubject.onNext(x);
            accessor.value = x;
            accessor.changedSubject.onNext(x);
        }, x=> accessor.thrownExceptions.onNext(x));

    return accessor;
}

RxExtension.d.ts

declare module Rx {
    export interface Observable<T> extends IObservable<T> {
        toProperty(initialValue?: T): wx.IObservableProperty<T>;
    }
}

如何在 TypeScript 中创建扩展 Rx.Observable 的 class？

How can I create a class that extends Rx.Observable in TypeScript?

rxjs

typescript