Observable.on订阅RxJs中的等价物

Question

在RxJava中,有Observable.OnSubscribe,当订阅了observable时会调用你的动作.

RxJs中有等价物吗？

---

自从提出这个问题以来,他们似乎已经把页面放下了.谷歌的缓存版本就在这里.

Answer 1

通过组合RxJs5的现有功能,可以轻松实现此功能.

Rx.Observable.prototype.doOnSubscribe = function (onSubscribe) {
  return Rx.Observable.empty()
    .do(null,null, onSubscribe)
    .concat(this);
};

Rx.Observable.from([1,2,3])
  .doOnSubscribe(() => console.log('subscribed to stream'))
  .subscribe(console.log, null, () => console.log('completed'))

<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/5.0.3/Rx.js"></script>

当下游订阅时,doOnSubscribe我们首先发出一个空的observable并使用.do()onComplete回调作为doOnSubscribe.然后我们.concat()上游到此并返回此修改后的流.

Answer 2

确实很像Mark的答案，但是如果您使用的是rxjs版本5或更高版本，我宁愿使用纯函数而不是修补原型。

更新：

按照Ray的建议empty concat使用defer，无需破解即可实现

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)
        );
    };
}

Answer 3

另一个以Mark的答案为基础的例子是使用工厂函数:

Rx.Observable.prototype.doOnSubscribe = function(onSubscribe) {
  const source = this; 
  return Rx.Observable.defer(() => {
      onSubscribe();
      return source;
  });
}