rxjs - RxJ 中的 Observable.onSubscribe 等效项

Question

在RxJava中，有Observable.OnSubscribe当有东西订阅可观察对象时，它将调用您的操作。

RxJs 中有等效的吗？

<小时/>

自从提出这个问题以来，他们似乎已经删除了该页面。 Google的缓存版本lives here .

Answer 1

非常喜欢 Mark 的回答，但如果您使用的是 rxjs 版本 5 或更高版本，我更喜欢使用纯函数而不是修补原型(prototype)。

更新:

按照 Ray 的建议，使用 defer 实现无 empty concat hack

import {defer} from 'rxjs/observable/defer';
import {Observable} from 'rxjs/Observable';

/** Example
import {from} from 'rxjs/observable/from';

from([1, 2, 3])
    .pipe(doOnSubscribe(() => console.log('subscribed to stream')))
    .subscribe(x => console.log(x), null, () => console.log('completed'));
*/

export function doOnSubscribe<T>(onSubscribe: () => void): (source: Observable<T>) =>  Observable<T> {
    return function inner(source: Observable<T>): Observable<T> {
        return defer(() => {
          onSubscribe();
          return source;
        });
    };
}

https://gist.github.com/evxn/750702f7c8e8d5a32c7b53167fe14d8d

原始答案:

import {empty} from 'rxjs/observable/empty';
import {concat, tap} from 'rxjs/operators';
import {Observable} from 'rxjs/Observable';

/** Example
import {from} from 'rxjs/observable/from';
from([1, 2, 3])
    .pipe(doOnSubscribe(() => console.log('subscribed to stream')))
    .subscribe(x => console.log(x), null, () => console.log('completed'));
*/

export function doOnSubscribe<T>(callback: () => void): (source: Observable<T>) =>  Observable<T> {
    return function inner(source: Observable<T>): Observable<T> {
        return empty().pipe(
            tap(null, null, callback),
            concat<T>(source)
        );
    };
}