Proposal: Drain operator

[vitaly-t]

In our company, we found the following generic operator being quite useful:

/**
 * Drains the source observable till it completes, and then posts an optional value-observable.
 */
function drain<T>(value?: T | Observable<T> | (() => T | Observable<T>)) {
    const v = () => {
        const a = typeof value === 'function' ? value.call(null) : value;
        return a instanceof Observable ? a : of(a);
    }
    return s => defer(() => s.pipe(filter(_ => false), c => concat(c, v()))) as Observable<T>;
}

Usage Examples

When we want to measure how long it takes a finite stream to complete, we use the following helper:

const startTimer = () => drain(Date.now);

so when we want to measure any particular stream, we can just add this operator, and then log the result:

obs.pipe(take(1000), startTimer())
   .subscribe(start => {
       // quickly check if this stream presents a bottleneck,
       // log how long it took to finish streaming 1000 values:
       console.log(Date.now() - start); 
   });

Other examples are when we want source observable drained first, and then a new observable created:

obs.pipe(take(1000), drain(from([1, 2, 3])))
   .subscribe(value => {
        // 1, 2, 3
   });

I wish such drain operator were part of RXJS ;)