fix(SafeSubscriber): Propagate disposal to terminal Subscriber

Ensure the internal SafeSubscriber is disposed when an Observable is subscribed with a Subscriber or Subscriber-like object.

Fixes ReactiveX#2675

spec/Observable-spec.ts

@@ -599,6 +599,45 @@ describe('Observable', () => {
         console.log = _log;
       });
     });
+
+    it('should fully propogate disposal when subscribed with a Subscriber', () => {
+      let observableUnsubscribed = false;
+      let subscriberUnsubscribed = false;
+      let subscriptionUnsubscribed = false;
+
+      const subscriber = new Subscriber<any>();
+      // verify unsubscribe is called on the Subscriber
+      subscriber.add(() => subscriberUnsubscribed = true);
+      const source = new Observable(_ => () => observableUnsubscribed = true);
+      const subscription = source.subscribe(subscriber);
+      // verify unsubscribe is called on children of the returned Subscription
+      subscription.add(() => subscriptionUnsubscribed = true);
+
+      subscription.unsubscribe();
+
+      expect(observableUnsubscribed).to.be.true;
+      expect(subscriberUnsubscribed).to.be.true;
+      expect(subscriptionUnsubscribed).to.be.true;
+    });
+
+    it('should fully propogate disposal when subscribed with a Subscriber-like', () => {
+      let observableUnsubscribed = false;
+      let subscriberUnsubscribed = false;
+      let subscriptionUnsubscribed = false;
+
+      // verify unsubscribe is called on the "Subscriber" (quack-quack)
+      const subscriber = { unsubscribe: () => subscriberUnsubscribed = true };
+      const source = new Observable(_ => () => observableUnsubscribed = true);
+      const subscription = source.subscribe(<any> subscriber as Subscriber<any>);
+      // verify unsubscribe is called on children of the returned Subscription
+      subscription.add(() => subscriptionUnsubscribed = true);
+
+      subscription.unsubscribe();
+
+      expect(observableUnsubscribed).to.be.true;
+      expect(subscriberUnsubscribed).to.be.true;
+      expect(subscriptionUnsubscribed).to.be.true;
+    });
   });
 
   describe('pipe', () => {

src/internal/Subscriber.ts

@@ -77,6 +77,7 @@ export class Subscriber<T> extends Subscription implements Observer<T> {
           } else {
             this.syncErrorThrowable = true;
             this.destination = new SafeSubscriber<T>(this, <PartialObserver<any>> destinationOrNext);
+            this.add(<SafeSubscriber<T>> this.destination);
           }
           break;
         }

src/internal/util/toSubscriber.ts

@@ -18,8 +18,11 @@ export function toSubscriber<T>(
     }
   }
 
-  if (!nextOrObserver && !error && !complete) {
-    return new Subscriber(emptyObserver);
+  if (!error && !complete) {
+    if (!nextOrObserver) {
+      return new Subscriber(emptyObserver);
+    }
+    return new Subscriber(nextOrObserver);
   }
 
   return new Subscriber(nextOrObserver, error, complete);