Merge branch 'master' into ws-fix

README.md

@@ -27,7 +27,7 @@ By contributing or commenting on issues in this repository, whether you've read
 ### ES6 via npm
 
 ```sh
-npm install rxjs-es
+npm install rxjs
 ```
 
 To import the entire core set of functionality:

spec/observables/IteratorObservable-spec.ts

@@ -1,5 +1,6 @@
 import {expect} from 'chai';
 import * as Rx from '../../dist/cjs/Rx';
+import {queue} from '../../dist/cjs/scheduler/queue';
 import {IteratorObservable} from '../../dist/cjs/observable/IteratorObservable';
 
 declare const expectObservable;
@@ -46,6 +47,68 @@ describe('IteratorObservable', () => {
  );
  });
 
+ it('should finalize generators if the subscription ends', () => {
+ const iterator = {
+ finalized: false,
+ next() {
+ return { value: 'duck', done: false };
+ },
+ return() {
+ this.finalized = true;
+ }
+ };
+
+ const iterable = {
+ [Rx.Symbol.iterator]() {
+ return iterator;
+ }
+ };
+
+ const results = [];
+
+ IteratorObservable.create(iterable)
+ .take(3)
+ .subscribe(
+ x => results.push(x),
+ null,
+ () => results.push('GOOSE!')
+ );
+
+ expect(results).to.deep.equal(['duck', 'duck', 'duck', 'GOOSE!']);
+ expect(iterator.finalized).to.be.true;
+ });
+
+ it('should finalize generators if the subscription and it is scheduled', () => {
+ const iterator = {
+ finalized: false,
+ next() {
+ return { value: 'duck', done: false };
+ },
+ return() {
+ this.finalized = true;
+ }
+ };
+
+ const iterable = {
+ [Rx.Symbol.iterator]() {
+ return iterator;
+ }
+ };
+
+ const results = [];
+
+ IteratorObservable.create(iterable, queue)
+ .take(3)
+ .subscribe(
+ x => results.push(x),
+ null,
+ () => results.push('GOOSE!')
+ );
+
+ expect(results).to.deep.equal(['duck', 'duck', 'duck', 'GOOSE!']);
+ expect(iterator.finalized).to.be.true;
+ });
+
  it('should emit members of an array iterator on a particular scheduler', () => {
  const source = IteratorObservable.create(
  [10, 20, 30, 40],

src/observable/IteratorObservable.ts

@@ -36,6 +36,9 @@ export class IteratorObservable<T> extends Observable<T> {
  state.index = index + 1;
 
  if (subscriber.closed) {
+ if (typeof iterator.return === 'function') {
+ iterator.return();
+ }
  return;
  }
 
@@ -71,6 +74,9 @@ export class IteratorObservable<T> extends Observable<T> {
  subscriber.next(result.value);
  }
  if (subscriber.closed) {
+ if (typeof iterator.return === 'function') {
+ iterator.return();
+ }
  break;
  }
  } while (true);

src/operator/do.ts

@@ -48,7 +48,7 @@ import { TeardownLogic } from '../Subscription';
  * @owner Observable
  */
 /* tslint:disable:max-line-length */
-export function _do<T>(next: (x: T) => void, error?: (e: any) => void, complete?: (this: Observable<T>) => void): Observable<T>;
+export function _do<T>(this: Observable<T>, next: (x: T) => void, error?: (e: any) => void, complete?: () => void): Observable<T>;
 export function _do<T>(this: Observable<T>, observer: PartialObserver<T>): Observable<T>;
 /* tslint:disable:max-line-length */
 export function _do<T>(this: Observable<T>, nextOrObserver?: PartialObserver<T> | ((x: T) => void),