publishLast.ts
2.06 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
import { Observable } from '../Observable';
import { AsyncSubject } from '../AsyncSubject';
import { multicast } from './multicast';
import { ConnectableObservable } from '../observable/ConnectableObservable';
import { UnaryFunction } from '../types';
/**
* Returns a connectable observable sequence that shares a single subscription to the
* underlying sequence containing only the last notification.
*
* 
*
* Similar to {@link publish}, but it waits until the source observable completes and stores
* the last emitted value.
* Similarly to {@link publishReplay} and {@link publishBehavior}, this keeps storing the last
* value even if it has no more subscribers. If subsequent subscriptions happen, they will
* immediately get that last stored value and complete.
*
* ## Example
*
* ```ts
* import { interval } from 'rxjs';
* import { publishLast, tap, take } from 'rxjs/operators';
*
* const connectable =
* interval(1000)
* .pipe(
* tap(x => console.log("side effect", x)),
* take(3),
* publishLast());
*
* connectable.subscribe(
* x => console.log( "Sub. A", x),
* err => console.log("Sub. A Error", err),
* () => console.log( "Sub. A Complete"));
*
* connectable.subscribe(
* x => console.log( "Sub. B", x),
* err => console.log("Sub. B Error", err),
* () => console.log( "Sub. B Complete"));
*
* connectable.connect();
*
* // Results:
* // "side effect 0"
* // "side effect 1"
* // "side effect 2"
* // "Sub. A 2"
* // "Sub. B 2"
* // "Sub. A Complete"
* // "Sub. B Complete"
* ```
*
* @see {@link ConnectableObservable}
* @see {@link publish}
* @see {@link publishReplay}
* @see {@link publishBehavior}
*
* @return {ConnectableObservable} An observable sequence that contains the elements of a
* sequence produced by multicasting the source sequence.
* @method publishLast
* @owner Observable
*/
export function publishLast<T>(): UnaryFunction<Observable<T>, ConnectableObservable<T>> {
return (source: Observable<T>) => multicast(new AsyncSubject<T>())(source);
}