<commit_before>import {expect} from 'chai';
import * as Rx from '../dist/cjs/Rx';
import {TeardownLogic} from '../dist/cjs/Subscription';

declare const {asDiagram, expectObservable, rxTestScheduler};
const Subscriber = Rx.Subscriber;
const Observable = Rx.Observable;

declare const __root__: any;

function expectFullObserver(val) {
  expect(val).to.be.a('object');
  expect(val.next).to.be.a('function');
  expect(val.error).to.be.a('function');
  expect(val.complete).to.be.a('function');
  expect(val.closed).to.be.a('boolean');
}

/** @test {Observable} */
describe('Observable', () => {
  it('should be constructed with a subscriber function', (done: MochaDone) => {
    const source = new Observable(function (observer) {
      expectFullObserver(observer);
      observer.next(1);
      observer.complete();
    });

    source.subscribe(function (x) { expect(x).to.equal(1); }, null, done);
  });

  describe('forEach', () => {
    it('should iterate and return a Promise', (done: MochaDone) => {
      const expected = [1, 2, 3];
      const result = Observable.of(1, 2, 3).forEach(function (x) {
        expect(x).to.equal(expected.shift());
      }, Promise)
      .then(() => {
        done();
      });

      expect(result.then).to.be.a('function');
    });

    it('should reject promise when in error', (done: MochaDone) => {
      Observable.throw('bad').forEach((x: any) => {
        done(new Error('should not be called'));
      }, Promise).then(() => {
        done(new Error('should not complete'));
      }, (err: any) => {
        expect(err).to.equal('bad');
        done();
      });
    });

    it('should allow Promise to be globally configured', (done: MochaDone) => {
      let wasCalled = false;

      __root__.Rx = {};
      __root__.Rx.config = {};
      __root__.Rx.config.Promise = function MyPromise(callback) {
        wasCalled = true;
        return new Promise(callback);
      };

      Observable.of(42).forEach((x: number) => {
        expect(x).to.equal(42);
      }).then(() => {
        expect(wasCalled).to.be.true;
        delete __root__.Rx;
        done();
      });
    });

    it('should reject promise if nextHandler throws', (done: MochaDone) => {
      const results = [];

      Observable.of(1, 2, 3).forEach((x: number) => {
        if (x === 3) {
          throw new Error('NO THREES!');
        }
        results.push(x);
      }, Promise)
      .then(() => {
        done(new Error('should not be called'));
      }, (err) => {
        expect(err).to.be.an('error', 'NO THREES!');
        expect(results).to.deep.equal([1, 2]);
      }).then(() => {
        done();
      });
    });

    it('should handle a synchronous throw from the next handler and tear down', (done: MochaDone) => {
      const expected = new Error('I told, you <NAME>, twos are the debil!');
      let unsubscribeCalled = false;
      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
        observer.next(1);
        observer.next(2);
        observer.next(3);

        return () => {
          unsubscribeCalled = true;
        };
      });

      const results = [];
      syncObservable.forEach((x) => {
        results.push(x);
        if (x === 2) {
          throw expected;
        }
      }).then(
        () => {
          done(new Error('should not be called'));
        },
        (err) => {
          results.push(err);
          expect(results).to.deep.equal([1, 2, expected]);
          expect(unsubscribeCalled).to.be.true;
          done();
        });
    });

    it('should handle an asynchronous throw from the next handler and tear down', (done: MochaDone) => {
      const expected = new Error('I told, you <NAME>, twos are the debil!');
      let unsubscribeCalled = false;
      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
        let i = 1;
        const id = setInterval(() => observer.next(i++), 1);

        return () => {
          clearInterval(id);
          unsubscribeCalled = true;
        };
      });

      const results = [];
      syncObservable.forEach((x) => {
        results.push(x);
        if (x === 2) {
          throw expected;
        }
      }).then(
        () => {
          done(new Error('should not be called'));
        },
        (err) => {
          results.push(err);
          expect(results).to.deep.equal([1, 2, expected]);
          expect(unsubscribeCalled).to.be.true;
          done();
        });
    });
  });

  describe('subscribe', () => {
    it('should be synchronous', () => {
      let subscribed = false;
      let nexted;
      let completed;
      const source = new Observable((observer: Rx.Observer<string>) => {
        subscribed = true;
        observer.next('wee');
        expect(nexted).to.equal('wee');
        observer.complete();
        expect(completed).to.be.true;
      });

      expect(subscribed).to.be.false;

      let mutatedByNext = false;
      let mutatedByComplete = false;

      source.subscribe((x: string) => {
        nexted = x;
        mutatedByNext = true;
      }, null, () => {
        completed = true;
        mutatedByComplete = true;
      });

      expect(mutatedByNext).to.be.true;
      expect(mutatedByComplete).to.be.true;
    });

    it('should work when subscribe is called with no arguments', () => {
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        subscriber.next('foo');
        subscriber.complete();
      });

      source.subscribe();
    });

    it('should run unsubscription logic when an error is sent synchronously and subscribe is called with no arguments', () => {
      let unsubscribeCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        subscriber.error(0);
        return () => {
          unsubscribeCalled = true;
        };
      });

      try {
        source.subscribe();
      } catch (e) {
        // error'ing to an empty Observer re-throws, so catch and ignore it here.
      }

      expect(unsubscribeCalled).to.be.true;
    });

    it('should run unsubscription logic when an error is sent asynchronously and subscribe is called with no arguments', (done: MochaDone) => {
      let unsubscribeCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        const id = setInterval(() => {
          try {
            subscriber.error(0);
          } catch (e) {
            // asynchronously error'ing to an empty Observer re-throws, so catch and ignore it here.
          }
        }, 1);<commit_msg>test(Observable): add test case for subscribe calls with same observer<commit_after>import {expect} from 'chai';
import * as Rx from '../dist/cjs/Rx';
import {TeardownLogic} from '../dist/cjs/Subscription';

declare const {asDiagram, expectObservable, rxTestScheduler};
const Subscriber = Rx.Subscriber;
const Observable = Rx.Observable;

declare const __root__: any;

function expectFullObserver(val) {
  expect(val).to.be.a('object');
  expect(val.next).to.be.a('function');
  expect(val.error).to.be.a('function');
  expect(val.complete).to.be.a('function');
  expect(val.closed).to.be.a('boolean');
}

/** @test {Observable} */
describe('Observable', () => {
  it('should be constructed with a subscriber function', (done: MochaDone) => {
    const source = new Observable(function (observer) {
      expectFullObserver(observer);
      observer.next(1);
      observer.complete();
    });

    source.subscribe(function (x) { expect(x).to.equal(1); }, null, done);
  });

  describe('forEach', () => {
    it('should iterate and return a Promise', (done: MochaDone) => {
      const expected = [1, 2, 3];
      const result = Observable.of(1, 2, 3).forEach(function (x) {
        expect(x).to.equal(expected.shift());
      }, Promise)
      .then(() => {
        done();
      });

      expect(result.then).to.be.a('function');
    });

    it('should reject promise when in error', (done: MochaDone) => {
      Observable.throw('bad').forEach((x: any) => {
        done(new Error('should not be called'));
      }, Promise).then(() => {
        done(new Error('should not complete'));
      }, (err: any) => {
        expect(err).to.equal('bad');
        done();
      });
    });

    it('should allow Promise to be globally configured', (done: MochaDone) => {
      let wasCalled = false;

      __root__.Rx = {};
      __root__.Rx.config = {};
      __root__.Rx.config.Promise = function MyPromise(callback) {
        wasCalled = true;
        return new Promise(callback);
      };

      Observable.of(42).forEach((x: number) => {
        expect(x).to.equal(42);
      }).then(() => {
        expect(wasCalled).to.be.true;
        delete __root__.Rx;
        done();
      });
    });

    it('should reject promise if nextHandler throws', (done: MochaDone) => {
      const results = [];

      Observable.of(1, 2, 3).forEach((x: number) => {
        if (x === 3) {
          throw new Error('NO THREES!');
        }
        results.push(x);
      }, Promise)
      .then(() => {
        done(new Error('should not be called'));
      }, (err) => {
        expect(err).to.be.an('error', 'NO THREES!');
        expect(results).to.deep.equal([1, 2]);
      }).then(() => {
        done();
      });
    });

    it('should handle a synchronous throw from the next handler and tear down', (done: MochaDone) => {
      const expected = new Error('I told, you <NAME>, twos are the debil!');
      let unsubscribeCalled = false;
      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
        observer.next(1);
        observer.next(2);
        observer.next(3);

        return () => {
          unsubscribeCalled = true;
        };
      });

      const results = [];
      syncObservable.forEach((x) => {
        results.push(x);
        if (x === 2) {
          throw expected;
        }
      }).then(
        () => {
          done(new Error('should not be called'));
        },
        (err) => {
          results.push(err);
          expect(results).to.deep.equal([1, 2, expected]);
          expect(unsubscribeCalled).to.be.true;
          done();
        });
    });

    it('should handle an asynchronous throw from the next handler and tear down', (done: MochaDone) => {
      const expected = new Error('I told, you <NAME>, twos are the debil!');
      let unsubscribeCalled = false;
      const syncObservable = new Observable<number>((observer: Rx.Observer<number>) => {
        let i = 1;
        const id = setInterval(() => observer.next(i++), 1);

        return () => {
          clearInterval(id);
          unsubscribeCalled = true;
        };
      });

      const results = [];
      syncObservable.forEach((x) => {
        results.push(x);
        if (x === 2) {
          throw expected;
        }
      }).then(
        () => {
          done(new Error('should not be called'));
        },
        (err) => {
          results.push(err);
          expect(results).to.deep.equal([1, 2, expected]);
          expect(unsubscribeCalled).to.be.true;
          done();
        });
    });
  });

  describe('subscribe', () => {
    it('should be synchronous', () => {
      let subscribed = false;
      let nexted;
      let completed;
      const source = new Observable((observer: Rx.Observer<string>) => {
        subscribed = true;
        observer.next('wee');
        expect(nexted).to.equal('wee');
        observer.complete();
        expect(completed).to.be.true;
      });

      expect(subscribed).to.be.false;

      let mutatedByNext = false;
      let mutatedByComplete = false;

      source.subscribe((x: string) => {
        nexted = x;
        mutatedByNext = true;
      }, null, () => {
        completed = true;
        mutatedByComplete = true;
      });

      expect(mutatedByNext).to.be.true;
      expect(mutatedByComplete).to.be.true;
    });

    it('should work when subscribe is called with no arguments', () => {
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        subscriber.next('foo');
        subscriber.complete();
      });

      source.subscribe();
    });

    it('should work when subscribe is called with no arguments and other subscription with no arguments completes', (done: MochaDone) => {
      let nextCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        const id = setTimeout(() => {
          subscriber.next('foo');
          subscriber.complete();
          nextCalled = true;
        }, 1);
        return () => {
          clearTimeout(id);
        };
      });

      source.subscribe();

      Observable.empty().subscribe();

      setTimeout(() => {
        let err;
        let errHappened = false;
        try {
          expect(nextCalled).to.be.true;
        } catch (e) {
          err = e;
          errHappened = true;
        } finally {
          if (!errHappened) {
            done();
          } else {
            done(err);
          }
        }
      }, 100);
    });

    it('should work when subscribe is called with observer and other subscription with same observer instance completes', (done: MochaDone) => {
      let nextCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        const id = setTimeout(() => {
          subscriber.next('foo');
          subscriber.complete();
          nextCalled = true;
        }, 1);
        return () => {
          clearTimeout(id);
        };
      });

      let observer = {
        next: function () { /*noop*/ }
      };

      source.subscribe(observer);

      Observable.empty().subscribe(observer);

      setTimeout(() => {
        let err;
        let errHappened = false;
        try {
          expect(nextCalled).to.be.true;
        } catch (e) {
          err = e;
          errHappened = true;
        } finally {
          if (!errHappened) {
            done();
          } else {
            done(err);
          }
        }
      }, 100);
    });

    it('should run unsubscription logic when an error is sent synchronously and subscribe is called with no arguments', () => {
      let unsubscribeCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        subscriber.error(0);
        return () => {
          unsubscribeCalled = true;
        };
      });

      try {
        source.subscribe();
      } catch (e) {
        // error'ing to an empty Observer re-throws, so catch and ignore it here.
      }

      expect(unsubscribeCalled).to.be.true;
    });

    it('should run unsubscription logic when an error is sent asynchronously and subscribe is called with no arguments', (done: MochaDone) => {
      let unsubscribeCalled = false;
      const source = new Observable((subscriber: Rx.Subscriber<string>) => {
        const id = setInterval(() => {
          try {
            subscriber.error(0);
          } catch (e) {
            // asynchronously error'ing to an empty Observer re-throws, so catch and ignore it here.
          }
        }, 1);
{'fraction_non_alphanumeric': 0.15633886647708484, 'fraction_numerical': 0.00476158984194167, 'mean_word_length': 2.1102426984779927, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 121, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6402586', 'n_tokens_mistral': 4487, 'n_tokens_neox': 4337, 'n_words': 1045}