如何在 RxJS 6 管道运算符中保留一个值？

Question

我想获取一个 JavaScript/TypeScript 对象，在 HTML 页面上构建一个元素，添加一个 CSS 类以使其显示，然后在一段时间后删除该类以将其从显示中删除。它在带有setTimeout 的subscribe 回调中运行良好。

of(recordObj).subscribe(
    (record) => {
        let recordEl = new MyRecord(record)
        console.log('created recordEl ' + recordEl.id);
        recordEl.show();
        setTimeout(_ => {
            console.log('hiding recordEl ' + recordEl.id);
            recordEl.hide();
        }, 4000);
    },
    _ => { console.log('subscribe error', _); },
    () => { console.log('done'); }
);

而不是setTimeout，我真的很想使用delay 使它成为一个RxJS 6 管道运算符。但是，我无法创建一个跨管道运算符持续存在的变量并且的范围仅限于给定记录通过管道。

在以下所有四次尝试中，recordEl 仅保留最近的记录。最后一个tap 在每个管道中遇到该记录时重复该记录的操作，或者，如果启用了空分配，则在catch[2] 处引发错误。

export function displayRecord (hideDelay: number = 5000, recordEl?: MyRecord /* attempt 1 */): OperatorFunction<MyRecord, MyRecord> {
    return (source: Observable<MyRecord>) => {
        let recordEl: MyRecord; // attempt 2
        // return ((recordEl?: MyRecord) => { // attempt 3
            // let recordEl: MyRecord; // attempt 4

            return source.pipe(
                tap((recordObj) => {
                    // create the element
                    recordEl = new MyRecord(recordObj);
                    console.log('created recordEl ' + recordEl.id);
                    recordEl.show();
                }),
                catchError((err, caught) => {
                    console.error('catch[1]', err);
                    return EMPTY;
                }),
                delay(hideDelay),
                tap((recordObj) => {
                    recordEl.hide(true);
                    console.log('destroying recordEl ' + recordEl.id);
                    // recordEl = null;
                    // console.log('destroyed recordEl ', recordEl);
                }),
                catchError((err, caught) => {
                    console.error('catch[2]', err);
                    return EMPTY;
                }),
            );
        // })();
    };
}

of(recordObj).pipe(
    displayRecord(4000),
).subscribe(
    () => {},
    _ => { console.log('subscribe error', _); },
    () => { console.log('done'); }
);

我在 SO 上找到了this question。尽管一个答案提到“如果 start$ Observable 发出更多值，它可能会中断”，但没有一个答案实际上解决了多个弹珠的情况。

这可能吗？怎么做？我是否在尝试反模式并且应该坚持使用subscribe 和setTimeout？

澄清：

我希望原始记录“通过”，最好不要将record 和recordEl 都通过管道。这个 pipe-of-pipes 使用 tap，因为它处理数据，而不是处理数据。

Answer 1

只需在处理过程中将您想要的任何信息传递到管道中即可。创建元素并将其插入到流中，延迟，然后从流中移除元素，让流看起来像它开始的样子。

类似这样的：

export function displayRecord (hideDelay: number = 5000): OperatorFunction<MyRecord, MyRecord> {
  return (source: Observable<MyRecord>) => 
    source.pipe(
      map(recordObj => {
        // create the element
        recordEl = new MyRecord(recordObj);
        console.log('created recordEl ' + recordEl.id);
        recordEl.show();
        return ({
          record: recordObj,
          element: recordEl
        });
      }),
      delay(hideDelay),
      map(rec => {
        rec.element.hide(true);
        console.log('destroying recordEl ' + rec.element.id);
        return rec.record;
      })
    );
}

Answer 2

我认为这将是另一种方法：

src$.pipe(
  mergeMap(recordObj => {
    let recordEl: MyRecord;

    return of(recordObj).pipe(
      tap(() => {
        recordEl = new MyRecord(recordObj);
        console.log('created recordEl ' + recordEl.id);
        recordEl.show();
      }),
      delay(/* ... */),
      tap(recordObj => {
        recordEl.hide();
      }),
    )
  }),
)

Answer 3

import {map, tap , delay } from 'rxjs/operators';

DELAY_TIME = 4000;
of(recordObj).pipe(
    map((record) => {
        let recordEl = new MyRecord(record)
        console.log('created recordEl ' + recordEl.id);
        recordEl.show();
        console.log('hiding recordEl ' + recordEl.id);
        return recordEl;
    }),
    delay(DELAY_TIME),
    tap((recordEl) => {
      recordEl.hide();
      console.log('recordEl got hidden' + recordEl.id);
    })
).subscribe();