Skip to content

Commit 61b877a

Browse files
authoredDec 15, 2022
feat(buffer): closingNotifier should support ObservableInput (#7073)
* feat(buffer): closingNotifier should support ObservableInput * chore(buffer): remove bad test
1 parent d501961 commit 61b877a

File tree

3 files changed

+49
-7
lines changed

3 files changed

+49
-7
lines changed
 

‎spec-dtslint/operators/buffer-spec.ts

+4
Original file line numberDiff line numberDiff line change
@@ -9,3 +9,7 @@ it('should enforce types', () => {
99
const o = of(1, 2, 3).pipe(buffer()); // $ExpectError
1010
const p = of(1, 2, 3).pipe(buffer(6)); // $ExpectError
1111
});
12+
13+
it('should support Promises', () => {
14+
const o = of(1, 2, 3).pipe(buffer(Promise.resolve('foo'))); // $ExpectType Observable<number[]>
15+
});

‎spec/operators/buffer-spec.ts

+38-1
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
import { buffer, mergeMap, take, window, toArray } from 'rxjs/operators';
2-
import { EMPTY, NEVER, throwError, of, Subject } from 'rxjs';
2+
import { EMPTY, NEVER, throwError, of, Subject, interval } from 'rxjs';
33
import { TestScheduler } from 'rxjs/testing';
44
import { observableMatcher } from '../helpers/observableMatcher';
55
import { expect } from 'chai';
@@ -324,6 +324,43 @@ describe('Observable.prototype.buffer', () => {
324324
expect(results).to.deep.equal([[1], [2], [], 'complete']);
325325
});
326326

327+
it('should buffer when Promise resolves', (done) => {
328+
const e1 = interval(3).pipe(take(5));
329+
const expected = [
330+
[0, 1],
331+
[2, 3, 4],
332+
];
333+
334+
e1.pipe(buffer(new Promise<void>((resolve) => setTimeout(() => resolve(), 8)))).subscribe({
335+
next: (x) => {
336+
expect(x).to.deep.equal(expected.shift());
337+
},
338+
error: () => done(new Error('should not be called')),
339+
complete: () => {
340+
expect(expected.length).to.equal(0);
341+
done();
342+
},
343+
});
344+
});
345+
346+
it('should raise error when Promise rejects', (done) => {
347+
const e1 = interval(1).pipe(take(5));
348+
const error = new Error('err');
349+
350+
e1.pipe(buffer(Promise.reject(error))).subscribe({
351+
next: () => {
352+
done(new Error('should not be called'));
353+
},
354+
error: (err: any) => {
355+
expect(err).to.be.an('error');
356+
done();
357+
},
358+
complete: () => {
359+
done(new Error('should not be called'));
360+
},
361+
});
362+
});
363+
327364
describe('equivalence with the window operator', () => {
328365
const cases = [
329366
{

‎src/internal/operators/buffer.ts

+7-6
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,8 @@
1-
import { Observable } from '../Observable';
2-
import { OperatorFunction } from '../types';
1+
import { OperatorFunction, ObservableInput } from '../types';
32
import { operate } from '../util/lift';
43
import { noop } from '../util/noop';
54
import { createOperatorSubscriber } from './OperatorSubscriber';
5+
import { innerFrom } from '../observable/innerFrom';
66

77
/**
88
* Buffers the source Observable values until `closingNotifier` emits.
@@ -13,7 +13,8 @@ import { createOperatorSubscriber } from './OperatorSubscriber';
1313
* ![](buffer.png)
1414
*
1515
* Buffers the incoming Observable values until the given `closingNotifier`
16-
* Observable emits a value, at which point it emits the buffer on the output
16+
* `ObservableInput` (that internally gets converted to an Observable)
17+
* emits a value, at which point it emits the buffer on the output
1718
* Observable and starts a new buffer internally, awaiting the next time
1819
* `closingNotifier` emits.
1920
*
@@ -36,12 +37,12 @@ import { createOperatorSubscriber } from './OperatorSubscriber';
3637
* @see {@link bufferWhen}
3738
* @see {@link window}
3839
*
39-
* @param {Observable<any>} closingNotifier An Observable that signals the
40+
* @param closingNotifier An `ObservableInput` that signals the
4041
* buffer to be emitted on the output Observable.
4142
* @return A function that returns an Observable of buffers, which are arrays
4243
* of values.
4344
*/
44-
export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T, T[]> {
45+
export function buffer<T>(closingNotifier: ObservableInput<any>): OperatorFunction<T, T[]> {
4546
return operate((source, subscriber) => {
4647
// The current buffered values.
4748
let currentBuffer: T[] = [];
@@ -59,7 +60,7 @@ export function buffer<T>(closingNotifier: Observable<any>): OperatorFunction<T,
5960
);
6061

6162
// Subscribe to the closing notifier.
62-
closingNotifier.subscribe(
63+
innerFrom(closingNotifier).subscribe(
6364
createOperatorSubscriber(
6465
subscriber,
6566
() => {

0 commit comments

Comments
 (0)
Please sign in to comment.