fix(IteratorObservable): get new iterator for each subscription

BREAKING CHANGE: IteratorObservable no longer share iterator between
subscription

- closes #2496

spec/observables/IteratorObservable-spec.ts

@@ -47,6 +47,24 @@ describe('IteratorObservable', () => {
       );
   });
 
+  it('should get new iterator for each subscription', () => {
+    const expected = [
+      Rx.Notification.createNext(10),
+      Rx.Notification.createNext(20),
+      Rx.Notification.createComplete()
+    ];
+
+    const e1 = IteratorObservable.create<number>(new Int32Array([10, 20])).observeOn(rxTestScheduler);
+
+    let v1, v2: Array<Rx.Notification<any>>;
+    e1.materialize().toArray().subscribe((x) => v1 = x);
+    e1.materialize().toArray().subscribe((x) => v2 = x);
+
+    rxTestScheduler.flush();
+    expect(v1).to.deep.equal(expected);
+    expect(v2).to.deep.equal(expected);
+  });
+
   it('should finalize generators if the subscription ends', () => {
     const iterator = {
       finalized: false,

src/observable/IteratorObservable.ts

@@ -11,8 +11,6 @@ import { Subscriber } from '../Subscriber';
  * @hide true
  */
 export class IteratorObservable<T> extends Observable<T> {
-  private iterator: any;
-
   static create<T>(iterator: any, scheduler?: IScheduler): IteratorObservable<T> {
     return new IteratorObservable(iterator, scheduler);
   }
@@ -45,20 +43,20 @@ export class IteratorObservable<T> extends Observable<T> {
     (<any> this).schedule(state);
   }
 
-  constructor(iterator: any, private scheduler?: IScheduler) {
+  constructor(private readonly iteratorObject: any, private scheduler?: IScheduler) {
     super();
 
-    if (iterator == null) {
+    if (iteratorObject == null) {
       throw new Error('iterator cannot be null.');
+    } else if (!iteratorObject[Symbol_iterator]) {
+      throw new TypeError('object is not iterable');
     }
-
-    this.iterator = getIterator(iterator);
   }
 
   protected _subscribe(subscriber: Subscriber<T>): TeardownLogic {
-
     let index = 0;
-    const { iterator, scheduler } = this;
+    const { scheduler } = this;
+    const iterator = getIterator(this.iteratorObject);
 
     if (scheduler) {
       return scheduler.schedule(IteratorObservable.dispatch, 0, {