123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296 |
- import { root } from './util/root';
- import { toSubscriber } from './util/toSubscriber';
- import { observable as Symbol_observable } from './symbol/observable';
- import { pipeFromArray } from './util/pipe';
- export class Observable {
-
- constructor(subscribe) {
- this._isScalar = false;
- if (subscribe) {
- this._subscribe = subscribe;
- }
- }
-
- lift(operator) {
- const observable = new Observable();
- observable.source = this;
- observable.operator = operator;
- return observable;
- }
-
- subscribe(observerOrNext, error, complete) {
- const { operator } = this;
- const sink = toSubscriber(observerOrNext, error, complete);
- if (operator) {
- operator.call(sink, this.source);
- }
- else {
- sink.add(this.source || !sink.syncErrorThrowable ? this._subscribe(sink) : this._trySubscribe(sink));
- }
- if (sink.syncErrorThrowable) {
- sink.syncErrorThrowable = false;
- if (sink.syncErrorThrown) {
- throw sink.syncErrorValue;
- }
- }
- return sink;
- }
- _trySubscribe(sink) {
- try {
- return this._subscribe(sink);
- }
- catch (err) {
- sink.syncErrorThrown = true;
- sink.syncErrorValue = err;
- sink.error(err);
- }
- }
-
- forEach(next, PromiseCtor) {
- if (!PromiseCtor) {
- if (root.Rx && root.Rx.config && root.Rx.config.Promise) {
- PromiseCtor = root.Rx.config.Promise;
- }
- else if (root.Promise) {
- PromiseCtor = root.Promise;
- }
- }
- if (!PromiseCtor) {
- throw new Error('no Promise impl found');
- }
- return new PromiseCtor((resolve, reject) => {
-
-
- let subscription;
- subscription = this.subscribe((value) => {
- if (subscription) {
-
-
-
-
- try {
- next(value);
- }
- catch (err) {
- reject(err);
- subscription.unsubscribe();
- }
- }
- else {
-
-
-
-
-
-
- next(value);
- }
- }, reject, resolve);
- });
- }
- _subscribe(subscriber) {
- return this.source.subscribe(subscriber);
- }
-
- [Symbol_observable]() {
- return this;
- }
-
-
- pipe(...operations) {
- if (operations.length === 0) {
- return this;
- }
- return pipeFromArray(operations)(this);
- }
-
- toPromise(PromiseCtor) {
- if (!PromiseCtor) {
- if (root.Rx && root.Rx.config && root.Rx.config.Promise) {
- PromiseCtor = root.Rx.config.Promise;
- }
- else if (root.Promise) {
- PromiseCtor = root.Promise;
- }
- }
- if (!PromiseCtor) {
- throw new Error('no Promise impl found');
- }
- return new PromiseCtor((resolve, reject) => {
- let value;
- this.subscribe((x) => value = x, (err) => reject(err), () => resolve(value));
- });
- }
- }
- Observable.create = (subscribe) => {
- return new Observable(subscribe);
- };
|