我想你要找的接线员是 throttle .

Stackblitz . 实现这一点的关键是设置传递给的配置对象 throttle() 这使得它能够排放(并处理)领先和落后的排放源,但在这段时间内忽略了两者之间的任何排放 processData()

以下是Stackblitz的关键功能:

// Use 'from' to emit the above array one object at a time
const source$ = from(sourceData).pipe(

  // Simulate a delay of 'delay * delayInterval' between emissions
  concatMap(data => of(data).pipe(delay(data.delay * delayInterval))),

  // Now tap in order to show the emissions on the console.
  tap(data => console.log('next ', data.emit)),

  // Finally, throttle as long as 'processData' is handling the emission
  throttle(data => processData(data), { leading: true, trailing: true }),

).subscribe()

更新:

上述代码的“一个问题”是,当源可观测对象完成时,

// Set up a Subject to be the source of data so we can manually complete it
const source$ = new Subject();

// the data observable is set up just to emit as per the gist.
const dataSubscribe = from(sourceData).pipe(
    // Simulate a delay of 'delay * delayInterval' before the emission
    concatMap(data => of(data).pipe(delay(data.delay * delayInterval))),
).subscribe(data => {
    console.log('next ', data.emit); // log the emission to console
    source$.next(data); // Send this emission into the source
});

// Finally, subscribe to the source$ so we can process the data
const sourceSubscribe = source$.pipe(
    // throttle as long as 'processData' is handling the emission
    throttle(data => processData(data), { leading: true, trailing: true })
).subscribe(); // will need to manually unsubscribe later ...

这是我能做出的最简单的解决方案:

const source = new Subject();
const start = new Date();
const mockDelayedObs = val => of(val).pipe(delay(1200));

source.pipe(
  multicast(
    new ReplaySubject(1),
    subject => {
      let lastValue;

      return subject.pipe(
        filter(v => v !== lastValue),
        exhaustMap(v => {
          lastValue = v;
          return mockDelayedObs(v);
        }),
        take(1),
        repeat(),
      );
    }
  ),
)
.subscribe(v => {
  console.log(new Date().getTime() - start.getTime(), v);
});

setTimeout(() => source.next(0), 0);
setTimeout(() => source.next(1), 500);
setTimeout(() => source.next(2), 1000);
setTimeout(() => source.next(3), 1500);
setTimeout(() => source.next(4), 1800);
setTimeout(() => source.next(5), 4000);

https://stackblitz.com/edit/rxjs-z33jgp?devtoolsheight=60

next 0
start handling 0
next 1
next 2
finish handling 0
start handling 2
next 3
next 4
finish handling 2
start handling 4
finish handling 4
start handling 5
finish handling 4

因此,只会打印0、2、4和5

这将在没有 multicast lastValue . 此变量仅用于忽略调用 mockDelayedObs 在重新订阅同一连锁店后,使用 repeat() .

也许你可以试着使用 race 方法论 b c 就在表演完之后 mergeMap 在…上 a ?

a.pipe(
  mergeMap(AResult => {
     // store aResult
     return race(b,c)
  }
).subscribe(
   finalResult => {
      // final result corresponding to either b or c
   }
)

A.

呸,这是一个很难破解的问题:

https://stackblitz.com/edit/angular-yk7akk

• 即时性是指可以立即执行的项目
• 延迟数据项基于lastFinished$。它将发出阻止执行的最后一项。

concatMap然后对这两个观测值进行合并。。

它的工作原理如前所述,但它并不是一个简单或直接的方法(命令式代码)。我正在关注这个讨论,寻找更优雅的解决方案。

@dmcgrandle解决方案重写为运算符:

export function trailingMap<T, O extends Observable<any>>(
  project: (value: T) => O
): OperatorFunction<T, ObservedValueOf<O>> {
  return (source: Observable<T>) => {
    const values$ = new Subject<any>();

    source
      .pipe(throttle(data => project(data)
        .pipe(
          tap(
            next => values$.next(next),
            e => values$.error(e)
          )
        ), {leading: true, trailing: true}))
      .subscribe(
        () => {},
        () => {},
        () => {
          values$.complete();
        },
      );

    return values$.asObservable();
  };
}

export function trailingMap<T, R, O extends ObservableInput<any>>(
  project: (value: T, index: number) => O,
): OperatorFunction<T, ObservedValueOf<O> | R> {
  // return mergeMap(project, 1);

    return multicast(
      new ReplaySubject(1),
      subject => {
        let lastValue;

        return subject.pipe(
          filter(v => v !== lastValue),
          exhaustMap(v => {
            lastValue = v;
            return project(value, index);
          }),
          take(1),
          repeat(),
        );
      }
    )
}

测试:

import { fakeAsync, tick } from '@angular/core/testing';
import { Observer, of, Subject } from 'rxjs';
import { delay, map } from 'rxjs/operators';

import { trailingMap } from '@shared/operators/trailing-map';

describe('trailingMap', () => {
  it('should skip intermediate source emissions', fakeAsync(() => {
    const source = new Subject();
    const mockDelayedObs = val =>
      of(val).pipe(
        delay(1200),
        map(value => `X${value}`)
      );

    const result = [];

    const observer: Observer<any> = {
      complete(): void {
        result.push('complete');
      },
      error(): void {},
      next(value: any): void {
        result.push(value);
      }
    };
    source
      .pipe(trailingMap(value => mockDelayedObs(value)))
      .subscribe(observer);

    // 0----1----2----3--4--------------------5|
    // \    x    \    x  \                    \
    // 0------------2------------4------------5|
    //     1200ms       1200ms       1200ms
    //

    setTimeout(() => source.next(0), 0);
    setTimeout(() => source.next(1), 500);
    setTimeout(() => source.next(2), 1000);
    setTimeout(() => source.next(3), 1500);
    setTimeout(() => source.next(4), 1800);
    setTimeout(() => source.next(5), 4000);
    setTimeout(() => source.complete(), 6000);

    tick(9000);
    expect(result).toEqual(['X0', 'X2', 'X4', 'X5', 'complete']);
  }));
});