包装一个可观察对象。在每个发出值之前和之后做一些事情

问题描述 投票：0回答：6

我想构建一个包装类，它在 Observable 的每个发出值之前和之后执行一些操作。

这是我的想法：

class Wrapper<T> {
    wrapped$: Observable<T>;

    _dataSubject = new Subject<T>();
    data$ = this._dataSubject.pipe(
        tap(_ => console.log("BEFORE"),
        //
        // map( ??? )
        //
    );

    constructor(wrapped$: Observable<T>) {
        this.wrapped$ = wrapped$.pipe(
            tap(_ => console.log("AFTER")
        );
    }
}

let subject = new Subject<string>();
let wrapper = new Wrapper(subject);
wrapper.data$.subscribe(val => console.log(val));
subject.next("foo")

控制台输出应该是：

BEFORE
foo
AFTER

我不知道如何将 

$wrapped
Observable 与 
_dataSubject
连接起来。

但也许我完全错了，需要采取不同的方法。

typescript rxjs observable rxjs5 subject-observer
6个回答
12
投票

这样的事情怎么样

import {Observable} from 'rxjs';

export class DoBeforeAfter<T> {
    wrapped$: Observable<T>;


    constructor(wrapped$: Observable<T>, doFunction: (data: any) => void) {
        this.wrapped$ = Observable.of(null)
            .do(_ => console.log("BEFORE"))
            .switchMap(_ => wrapped$)
            .do(doFunction)
            .do(_ => console.log('AFTER'));
    }

}

要这样吃

const source = Observable.of('NOW');
const beforeAfter = new DoBeforeAfter(source, data => console.log(data));

beforeAfter.wrapped$.subscribe(
        null,
        error => console.error(error),
        () => console.log('DONE')

）

看起来有点麻烦，但也许有帮助

4
投票

所以，根据我的理解，你可以这样做：

class Wrapper<T> {

    _dataSubject: Subject<T>;
    wrapped$: Observable<T>;

    constructor(wrapped$: Subject<T>) {
        this._dataSubject = wrapped$;
        this.wrapped$ = this._dataSubject.pipe(
          tap(_ => console.log("BEFORE")),
          tap(data => console.log(data)),
          tap(_ => console.log("AFTER"))
        );
    }
}

然后：

let subject = new Subject<string>();
let wrapper = new Wrapper(subject);
wrapper.wrapped$.subscribe();
subject.next("foo")
3
投票

最好的方法（虽然很复杂）是创建一个新的运算符，类似于 

tap
，但它会在值发出之前和之后执行一些操作。

您可以在示例中看到它的工作原理（它是在 ES6 中，因为 SO 代码片段不接受 TypeScript，但您会明白的）

function wrap(before, after) {
    return function wrapOperatorFunction(source) {
        return source.lift(new WrapOperator(before, after));
    };
}
class WrapOperator {
    constructor(before, after) {
        this.before = before;
        this.after = after;
    }
    call(subscriber, source) {
        return source.subscribe(new WrapSubscriber(subscriber, this.before, this.after));
    }
}

class WrapSubscriber extends Rx.Subscriber {
    constructor(destination, before, after) {
        super(destination);
        this.before = before;
        this.after = after;
    }
    _next(value) {
        this.before ? this.before(value) : null;
        this.destination.next(value);
        this.after ? this.after(value) : null;
    }
}

// Now:

const observable = Rx.Observable.from([1, 2, 3, 4]);

observable.pipe(
   wrap(value => console.log('before', value), value => console.log('after', value))
).subscribe(value => console.log('value emitted', value), null, () => console.log('complete'));


// For what you want:
// let's simulate that, for each value in the array, we'll fetch something from an external service:
// we want the before to be executed when we make the request, and the after to be executed when it finishes. In this // case, we just combine two wrap operators and flatMap, for example:

observable.pipe(
  wrap(value => console.log('BEFORE REQUEST', value)),
  Rx.operators.flatMap(value => {
    const subject = new Rx.Subject();
    setTimeout(() => { subject.next(value); subject.complete(); }, 5000);
    return subject;
  }),
  wrap(undefined, value => console.log('AFTER REQUEST', value))
).subscribe(value => console.log('value emitted', value), null, () => console.log('complete'));
<script src="https://unpkg.com/@reactivex/[email protected]/dist/global/Rx.js"></script>

如上所述，可能有点复杂，但它与 RxJS 运算符无缝集成，并且它始终是了解如何创建我们自己的运算符的一个很好的示例:-)

对于您在评论中所说的内容，您可以查看最后一个示例。在那里，我组合了两个 

wrap
运算符。第一个仅使用 
before
回调，因此它仅在发出值之前执行某些操作。如您所见，因为源可观察值来自数组，所以四个 
before
回调会立即执行。然后，我们应用
flatMap
。对此，我们应用一个新的 
wrap
，但这次仅使用 
after
回调。因此，只有在 
flatMap
返回的可观察量产生其值之后才会调用此回调。

当然，如果不是从数组中获取可观察值，而是从事件侦听器中创建一个可观察值，那么你将拥有：

  1. before
    回调在可观察对象推送事件之前执行。
  2. 异步可观察执行。
  3. 异步可观察量在时间 t 后产生值。
  4. 执行
    after
    回调。

这就是拥有运营商的回报所在，因为它们很容易组合。希望这适合您。

0
投票

感谢您的所有意见！

我现在想出了自己的解决方案。 

AFTER
和
DATA
输出的顺序是错误的，但我发现这并不重要，只要它们同时出现即可

代码仍然需要一些重构，但至少现在可以工作！

class Wrapper {
    _taskDoneSubject = new Subject<boolean>();
    taskDone$ = this._taskDoneSubject.asObservable();

    wrappedSwitchMap<T, R>(wrapped: (val: T) => Observable<R>): OperatorFunction<T, R> {
        return switchMap<T, R>(val => { // can also be done using mergeMap
            this._taskDoneSubject.next(false);
            return wrapped(val).pipe(
                tap(_ => this._taskDoneSubject.next(true))
            );
        });
    }
}

用途：

test(valueEmittingObservable: Observable<string>) {

    let wrapper = new Wrapper();
    wrapper.taskDone$.subscribe(val => console.log("task done? ", val);

    valueEmittingObservable.pipe(
        // wrapper.wrappedSwitchMap(val => anObservableThatTakesAWhile(val))
        wrapper.wrappedSwitchMap(val => of("foo").pipe(delay(5000)) // simulated
    ).subscribe(val => console.log("emitted value: ", val);

}

输出：

task done? false
(... 5 seconds delay ...)
task done? true
emitted value: foo

或者如果发射速度快于 5 秒：

task done? false
(... 1 second until next emit ...)
task done? false
(... 5 seconds delay ...)
task done? true
emitted value: foo
0
投票

如果您想在不使用 RxJs 6 的课程的情况下完成此操作：

return of(null)
    .pipe(tap(_ => {
        console.log('BEFORE');
    }))
    .pipe(switchMap(_ => wrappedObservable))
    .pipe(tap(_ => {
        console.log('AFTER');
    }))
0
投票

自 rxjs 7 起，使用新的 TapObeserver 输入的 tap 运算符现在有更多选项。 您现在可以围绕您的函数而无需创建包装类。

import { of } from 'rxjs';
import { delay, tap } from 'rxjs/operators';
import { MonoTypeOperatorFunction } from 'rxjs/interfaces';

function log<T>(): MonoTypeOperatorFunction<T> {
  return tap<T>({
    subscribe: () => console.log('Log before'),
    finalize: () => console.log('Log after')
  })
}

const load$ = of(42).pipe(delay(500));

load$.pipe(
  log()
).subscribe((v) => console.log('Emitted', v));


// Output:
// Log before
// Emitted 42
// Log after
最新问题
© www.soinside.com 2019 - 2024. All rights reserved.