如何在RxJS 6管道操作符中持久化一个值？

Question

我想要接受一个JavaScript/TypeScript对象，从它在HTML页面上构建一个元素，添加一个CSS类来使它显示，然后在某个时间段后删除这个类，以便从显示中删除它。它在subscribe回调和setTimeout中工作得很好。

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

我在上面找到了this question。虽然有一个答案提到“如果启动美元可观测到的价值更多，它就会崩溃”，但没有一个答案实际上解决了一个以上大理石的场景。

这有可能吗?怎么可能？我是否尝试了反模式，并且应该坚持subscribe和setTimeout？

CLARIFICATION:

我希望原始记录“失败”，最好不要通过管道同时传递record和recordEl。这个管道使用tap，因为它使用的是数据，而不是数据。

Answer 1

我认为这是另一种方法：

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 2

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

Answer 3

在处理管道时，只要把你想要的任何信息传递下去就行了。创建元素并将其插入流中，延迟，然后从流中删除元素，使流看起来像它开始的样子。

就像这样：

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