Использование observables для передачи значений¶

// Create an Observable that will start listening to geolocation updates
// when a consumer subscribes.
const locations = new Observable((observer) => {
    let watchId: number;

    // Simple geolocation API check provides values to publish
    if ('geolocation' in navigator) {
        watchId = navigator.geolocation.watchPosition(
            (position: GeolocationPosition) => {
                observer.next(position);
            },
            (error: GeolocationPositionError) => {
                observer.error(error);
            }
        );
    } else {
        observer.error('Geolocation not available');
    }

    // When the consumer unsubscribes, clean up data ready for next subscription.
    return {
        unsubscribe() {
            navigator.geolocation.clearWatch(watchId);
        },
    };
});

// Call subscribe() to start listening for updates.
const locationsSubscription = locations.subscribe({
    next(position) {
        console.log('Current Position: ', position);
    },
    error(msg) {
        console.log('Error Getting Location: ', msg);
    },
});

// Stop listening for location after 10 seconds
setTimeout(() => {
    locationsSubscription.unsubscribe();
}, 10000);

// Create simple observable that emits three values
const myObservable = of(1, 2, 3);

// Create observer object
const myObserver = {
    next: (x: number) =>
        console.log('Observer got a next value: ' + x),
    error: (err: Error) =>
        console.error('Observer got an error: ' + err),
    complete: () =>
        console.log('Observer got a complete notification'),
};

// Execute with the observer object
myObservable.subscribe(myObserver);

// Logs:
// Observer got a next value: 1
// Observer got a next value: 2
// Observer got a next value: 3
// Observer got a complete notification

myObservable.subscribe(
    (x) => console.log('Observer got a next value: ' + x),
    (err) => console.error('Observer got an error: ' + err),
    () =>
        console.log('Observer got a complete notification')
);

// This function runs when subscribe() is called
function sequenceSubscriber(observer: Observer<number>) {
    // synchronously deliver 1, 2, and 3, then complete
    observer.next(1);
    observer.next(2);
    observer.next(3);
    observer.complete();

    // unsubscribe function doesn't need to do anything in this
    // because values are delivered synchronously
    return { unsubscribe() {} };
}

// Create a new Observable that will deliver the above sequence
const sequence = new Observable(sequenceSubscriber);

// execute the Observable and print the result of each notification
sequence.subscribe({
    next(num) {
        console.log(num);
    },
    complete() {
        console.log('Finished sequence');
    },
});

// Logs:
// 1
// 2
// 3
// Finished sequence

function fromEvent<T extends keyof HTMLElementEventMap>(
    target: HTMLElement,
    eventName: T
) {
    return new Observable<HTMLElementEventMap[T]>(
        (observer) => {
            const handler = (e: HTMLElementEventMap[T]) =>
                observer.next(e);

            // Add the event handler to the target
            target.addEventListener(eventName, handler);

            return () => {
                // Detach the event handler from the target
                target.removeEventListener(
                    eventName,
                    handler
                );
            };
        }
    );
}

const ESC_CODE = 'Escape';
const nameInput = document.getElementById(
    'name'
) as HTMLInputElement;

const subscription = fromEvent(
    nameInput,
    'keydown'
).subscribe((e: KeyboardEvent) => {
    if (e.code === ESC_CODE) {
        nameInput.value = '';
    }
});

function sequenceSubscriber(observer: Observer<number>) {
    const seq = [1, 2, 3];
    let clearTimer: VoidFunction | undefined;

    // Will run through an array of numbers, emitting one value
    // per second until it gets to the end of the array.
    function doInSequence(arr: number[], idx: number) {
        const timeout = setTimeout(() => {
            observer.next(arr[idx]);
            if (idx === arr.length - 1) {
                observer.complete();
            } else {
                doInSequence(arr, ++idx);
            }
        }, 1000);
        clearTimer = () => clearTimeout(timeout);
    }

    doInSequence(seq, 0);

    // Unsubscribe should clear the timeout to stop execution
    return {
        unsubscribe() {
            clearTimer?.();
        },
    };
}

// Create a new Observable that will deliver the above sequence
const sequence = new Observable(sequenceSubscriber);

sequence.subscribe({
    next(num) {
        console.log(num);
    },
    complete() {
        console.log('Finished sequence');
    },
});

// Logs:
// (at 1 second): 1
// (at 2 seconds): 2
// (at 3 seconds): 3
// (at 3 seconds): Finished sequence

// Subscribe starts the clock, and will emit after 1 second
sequence.subscribe({
    next(num) {
        console.log('1st subscribe: ' + num);
    },
    complete() {
        console.log('1st sequence finished.');
    },
});

// After 1/2 second, subscribe again.
setTimeout(() => {
    sequence.subscribe({
        next(num) {
            console.log('2nd subscribe: ' + num);
        },
        complete() {
            console.log('2nd sequence finished.');
        },
    });
}, 500);

// Logs:
// (at 1 second): 1st subscribe: 1
// (at 1.5 seconds): 2nd subscribe: 1
// (at 2 seconds): 1st subscribe: 2
// (at 2.5 seconds): 2nd subscribe: 2
// (at 3 seconds): 1st subscribe: 3
// (at 3 seconds): 1st sequence finished
// (at 3.5 seconds): 2nd subscribe: 3
// (at 3.5 seconds): 2nd sequence finished

function multicastSequenceSubscriber() {
    const seq = [1, 2, 3];
    // Keep track of each observer (one for every active subscription)
    const observers: Observer<unknown>[] = [];
    // Still a single timer because there will only ever be one
    // set of values being generated, multicasted to each subscriber
    let clearTimer: VoidFunction | undefined;

    // Return the subscriber function (runs when subscribe()
    // function is invoked)
    return (observer: Observer<unknown>) => {
        observers.push(observer);
        // When this is the first subscription, start the sequence
        if (observers.length === 1) {
            const multicastObserver: Observer<number> = {
                next(val) {
                    // Iterate through observers and notify all subscriptions
                    observers.forEach((obs) =>
                        obs.next(val)
                    );
                },
                error() {
                    /* Handle the error... */
                },
                complete() {
                    // Notify all complete callbacks
                    observers
                        .slice(0)
                        .forEach((obs) => obs.complete());
                },
            };
            doSequence(multicastObserver, seq, 0);
        }

        return {
            unsubscribe() {
                // Remove from the observers array so it's no longer notified
                observers.splice(
                    observers.indexOf(observer),
                    1
                );
                // If there's no more listeners, do cleanup
                if (observers.length === 0) {
                    clearTimer?.();
                }
            },
        };

        // Run through an array of numbers, emitting one value
        // per second until it gets to the end of the array.
        function doSequence(
            sequenceObserver: Observer<number>,
            arr: number[],
            idx: number
        ) {
            const timeout = setTimeout(() => {
                console.log('Emitting ' + arr[idx]);
                sequenceObserver.next(arr[idx]);
                if (idx === arr.length - 1) {
                    sequenceObserver.complete();
                } else {
                    doSequence(
                        sequenceObserver,
                        arr,
                        ++idx
                    );
                }
            }, 1000);
            clearTimer = () => clearTimeout(timeout);
        }
    };
}

// Create a new Observable that will deliver the above sequence
const multicastSequence = new Observable(
    multicastSequenceSubscriber()
);

// Subscribe starts the clock, and begins to emit after 1 second
multicastSequence.subscribe({
    next(num) {
        console.log('1st subscribe: ' + num);
    },
    complete() {
        console.log('1st sequence finished.');
    },
});

// After 1 1/2 seconds, subscribe again (should "miss" the first value).
setTimeout(() => {
    multicastSequence.subscribe({
        next(num) {
            console.log('2nd subscribe: ' + num);
        },
        complete() {
            console.log('2nd sequence finished.');
        },
    });
}, 1500);

// Logs:
// (at 1 second): Emitting 1
// (at 1 second): 1st subscribe: 1
// (at 2 seconds): Emitting 2
// (at 2 seconds): 1st subscribe: 2
// (at 2 seconds): 2nd subscribe: 2
// (at 3 seconds): Emitting 3
// (at 3 seconds): 1st subscribe: 3
// (at 3 seconds): 2nd subscribe: 3
// (at 3 seconds): 1st sequence finished
// (at 3 seconds): 2nd sequence finished

myObservable.subscribe({
    next(num) {
        console.log('Next num: ' + num);
    },
    error(err) {
        console.log('Received an error: ' + err);
    },
});