1 | import { Operator } from '../Operator';
|
---|
2 | import { Observable } from '../Observable';
|
---|
3 | import { Subscriber } from '../Subscriber';
|
---|
4 | import { Subscription } from '../Subscription';
|
---|
5 |
|
---|
6 | import { Observer, OperatorFunction } from '../types';
|
---|
7 |
|
---|
8 | /**
|
---|
9 | * Compares all values of two observables in sequence using an optional comparator function
|
---|
10 | * and returns an observable of a single boolean value representing whether or not the two sequences
|
---|
11 | * are equal.
|
---|
12 | *
|
---|
13 | * <span class="informal">Checks to see of all values emitted by both observables are equal, in order.</span>
|
---|
14 | *
|
---|
15 | * ![](sequenceEqual.png)
|
---|
16 | *
|
---|
17 | * `sequenceEqual` subscribes to two observables and buffers incoming values from each observable. Whenever either
|
---|
18 | * observable emits a value, the value is buffered and the buffers are shifted and compared from the bottom
|
---|
19 | * up; If any value pair doesn't match, the returned observable will emit `false` and complete. If one of the
|
---|
20 | * observables completes, the operator will wait for the other observable to complete; If the other
|
---|
21 | * observable emits before completing, the returned observable will emit `false` and complete. If one observable never
|
---|
22 | * completes or emits after the other complets, the returned observable will never complete.
|
---|
23 | *
|
---|
24 | * ## Example
|
---|
25 | * figure out if the Konami code matches
|
---|
26 | * ```ts
|
---|
27 | * import { from, fromEvent } from 'rxjs';
|
---|
28 | * import { sequenceEqual, bufferCount, mergeMap, map } from 'rxjs/operators';
|
---|
29 | *
|
---|
30 | * const codes = from([
|
---|
31 | * 'ArrowUp',
|
---|
32 | * 'ArrowUp',
|
---|
33 | * 'ArrowDown',
|
---|
34 | * 'ArrowDown',
|
---|
35 | * 'ArrowLeft',
|
---|
36 | * 'ArrowRight',
|
---|
37 | * 'ArrowLeft',
|
---|
38 | * 'ArrowRight',
|
---|
39 | * 'KeyB',
|
---|
40 | * 'KeyA',
|
---|
41 | * 'Enter', // no start key, clearly.
|
---|
42 | * ]);
|
---|
43 | *
|
---|
44 | * const keys = fromEvent(document, 'keyup').pipe(map(e => e.code));
|
---|
45 | * const matches = keys.pipe(
|
---|
46 | * bufferCount(11, 1),
|
---|
47 | * mergeMap(
|
---|
48 | * last11 => from(last11).pipe(sequenceEqual(codes)),
|
---|
49 | * ),
|
---|
50 | * );
|
---|
51 | * matches.subscribe(matched => console.log('Successful cheat at Contra? ', matched));
|
---|
52 | * ```
|
---|
53 | *
|
---|
54 | * @see {@link combineLatest}
|
---|
55 | * @see {@link zip}
|
---|
56 | * @see {@link withLatestFrom}
|
---|
57 | *
|
---|
58 | * @param {Observable} compareTo The observable sequence to compare the source sequence to.
|
---|
59 | * @param {function} [comparator] An optional function to compare each value pair
|
---|
60 | * @return {Observable} An Observable of a single boolean value representing whether or not
|
---|
61 | * the values emitted by both observables were equal in sequence.
|
---|
62 | * @method sequenceEqual
|
---|
63 | * @owner Observable
|
---|
64 | */
|
---|
65 | export function sequenceEqual<T>(compareTo: Observable<T>,
|
---|
66 | comparator?: (a: T, b: T) => boolean): OperatorFunction<T, boolean> {
|
---|
67 | return (source: Observable<T>) => source.lift(new SequenceEqualOperator(compareTo, comparator));
|
---|
68 | }
|
---|
69 |
|
---|
70 | export class SequenceEqualOperator<T> implements Operator<T, boolean> {
|
---|
71 | constructor(private compareTo: Observable<T>,
|
---|
72 | private comparator: (a: T, b: T) => boolean) {
|
---|
73 | }
|
---|
74 |
|
---|
75 | call(subscriber: Subscriber<boolean>, source: any): any {
|
---|
76 | return source.subscribe(new SequenceEqualSubscriber(subscriber, this.compareTo, this.comparator));
|
---|
77 | }
|
---|
78 | }
|
---|
79 |
|
---|
80 | /**
|
---|
81 | * We need this JSDoc comment for affecting ESDoc.
|
---|
82 | * @ignore
|
---|
83 | * @extends {Ignored}
|
---|
84 | */
|
---|
85 | export class SequenceEqualSubscriber<T, R> extends Subscriber<T> {
|
---|
86 | private _a: T[] = [];
|
---|
87 | private _b: T[] = [];
|
---|
88 | private _oneComplete = false;
|
---|
89 |
|
---|
90 | constructor(destination: Observer<R>,
|
---|
91 | private compareTo: Observable<T>,
|
---|
92 | private comparator: (a: T, b: T) => boolean) {
|
---|
93 | super(destination);
|
---|
94 | (this.destination as Subscription).add(compareTo.subscribe(new SequenceEqualCompareToSubscriber(destination, this)));
|
---|
95 | }
|
---|
96 |
|
---|
97 | protected _next(value: T): void {
|
---|
98 | if (this._oneComplete && this._b.length === 0) {
|
---|
99 | this.emit(false);
|
---|
100 | } else {
|
---|
101 | this._a.push(value);
|
---|
102 | this.checkValues();
|
---|
103 | }
|
---|
104 | }
|
---|
105 |
|
---|
106 | public _complete(): void {
|
---|
107 | if (this._oneComplete) {
|
---|
108 | this.emit(this._a.length === 0 && this._b.length === 0);
|
---|
109 | } else {
|
---|
110 | this._oneComplete = true;
|
---|
111 | }
|
---|
112 | this.unsubscribe();
|
---|
113 | }
|
---|
114 |
|
---|
115 | checkValues() {
|
---|
116 | const { _a, _b, comparator } = this;
|
---|
117 | while (_a.length > 0 && _b.length > 0) {
|
---|
118 | let a = _a.shift();
|
---|
119 | let b = _b.shift();
|
---|
120 | let areEqual = false;
|
---|
121 | try {
|
---|
122 | areEqual = comparator ? comparator(a, b) : a === b;
|
---|
123 | } catch (e) {
|
---|
124 | this.destination.error(e);
|
---|
125 | }
|
---|
126 | if (!areEqual) {
|
---|
127 | this.emit(false);
|
---|
128 | }
|
---|
129 | }
|
---|
130 | }
|
---|
131 |
|
---|
132 | emit(value: boolean) {
|
---|
133 | const { destination } = this;
|
---|
134 | destination.next(value);
|
---|
135 | destination.complete();
|
---|
136 | }
|
---|
137 |
|
---|
138 | nextB(value: T) {
|
---|
139 | if (this._oneComplete && this._a.length === 0) {
|
---|
140 | this.emit(false);
|
---|
141 | } else {
|
---|
142 | this._b.push(value);
|
---|
143 | this.checkValues();
|
---|
144 | }
|
---|
145 | }
|
---|
146 |
|
---|
147 | completeB() {
|
---|
148 | if (this._oneComplete) {
|
---|
149 | this.emit(this._a.length === 0 && this._b.length === 0);
|
---|
150 | } else {
|
---|
151 | this._oneComplete = true;
|
---|
152 | }
|
---|
153 | }
|
---|
154 | }
|
---|
155 |
|
---|
156 | class SequenceEqualCompareToSubscriber<T, R> extends Subscriber<T> {
|
---|
157 | constructor(destination: Observer<R>, private parent: SequenceEqualSubscriber<T, R>) {
|
---|
158 | super(destination);
|
---|
159 | }
|
---|
160 |
|
---|
161 | protected _next(value: T): void {
|
---|
162 | this.parent.nextB(value);
|
---|
163 | }
|
---|
164 |
|
---|
165 | protected _error(err: any): void {
|
---|
166 | this.parent.error(err);
|
---|
167 | this.unsubscribe();
|
---|
168 | }
|
---|
169 |
|
---|
170 | protected _complete(): void {
|
---|
171 | this.parent.completeB();
|
---|
172 | this.unsubscribe();
|
---|
173 | }
|
---|
174 | }
|
---|