RXJS:向 Observable 添加一个函数以在订阅时执行(延迟) [英] RXJS: Adding a function to Observable to execute when subscribed to (defer)

问题描述

我有一个由事件组成的 Observable.在这种情况下，蓝牙通知.

I have an Observable made from events. In this case, Bluetooth notifications.

我只想在有人订阅该 Observable 时运行一个函数 (startNotifictions).

I want to run a function (startNotifictions) only when someone is subscribing to that Observable.

此代码在以前的版本中确实有效.它与 Ionic3 框架一起使用.它添加了一个新的运算符，在订阅时运行.现在转译器的类型有问题，抱怨两次，即 .doOnSubscribe 在 typedef Observable any> 上不可用和 <{}>.

This code did work, on previous versions. It is with Ionic3 framework. It added a new operator, that ran when subscribed. Now the transpiler has a problem with the types, complaining twice, that the .doOnSubscribe is not available on typedef Observable any> and <{}>.

有人知道如何正确输入吗?可以延长吗?试过直接用.defer，没用.

Anyone has an idea how to get that typed correctly? Extend maybe? Tried to use .defer directly, no avail.

 // add operator doOnSubscribe to the event observable
        Observable.prototype.doOnSubscribe = function(onSubscribe) {
            let source = this;
            return Observable.defer(() => {
                onSubscribe();
                return source;
            });
        };

        // return the Observable for the notify char, with startNotify on first subscribe
        getUartDataNote( Observable.fromEvent( this.uartChar, 'characteristicvaluechanged' )
            .doOnSubscribe(() => {
                console.log('starting note');
                this.uartChar.startNotifications();
            })
            .map( value => String.fromCharCode.apply( null, new Uint8Array( this.uartChar.value.buffer )))
            .takeUntil( Observable.fromEvent( this.gatt.device, 'gattserverdisconnected' ))
            .finally(() => {
                console.log( 'stream disconnected ');
                // not necessary: return this.uartChar.stopNotifications()
            })
            .share()
        );

推荐答案

下面是你写类型扩充的方法.

Here is how you write the type augmentation.

export {}

declare module 'rxjs/Observable' {
  interface Observable<T> {
    doOnSubscribe(onSubscribe: () => void): this;
  }
}

这在 声明合并 TypeScript 手册的部分.

This is documented in the Declaration Merging section of the TypeScript handbook.

这篇关于RXJS:向 Observable 添加一个函数以在订阅时执行(延迟)的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！