Make rxNewToRxNewAndOld work with old observers missing onNext.

Author: lrowe

src/run/conversion/rxNewToRxNewAndOld.js

@@ -1,57 +1,65 @@
+"use strict";
+function noop() {}
+
+function toRxNewObserver(observer) {
+    var onNext = observer.onNext;
+    var onError = observer.onError;
+    var onCompleted = observer.onCompleted;
+    if (
+        typeof onNext !== "function" &&
+        typeof onError !== "function" &&
+        typeof onCompleted !== "function"
+    ) {
+        return observer;
+    }
+    // old observer!
+    return {
+        next: typeof onNext === "function"
+            ? function(x) {
+                  this.destination.onNext(x);
+              }
+            : noop,
+        error: typeof onError === "function"
+            ? function(err) {
+                  this.destination.onError(err);
+              }
+            : noop,
+        complete: typeof onCompleted === "function"
+            ? function() {
+                  this.destination.onCompleted();
+              }
+            : noop,
+        destination: observer
+    };
+}
+
 // WHY NOT BOTH?
 module.exports = function rxNewToRxNewAndOld(rxNewObservable) {
-  var _subscribe = rxNewObservable.subscribe;
+    var _subscribe = rxNewObservable.subscribe;
 
-  rxNewObservable.subscribe = function (observerOrNextFn, errFn, compFn) {
-      var subscription;
-      switch (typeof observerOrNextFn) {
-          case 'function':
-              subscription = _subscribe.call(this,
-                  observerOrNextFn, errFn, compFn);
-              break;
-          case 'object':
-              var observer = observerOrNextFn;
-              if (typeof observerOrNextFn.onNext === 'function') {
-                  // old observer!
-                  observer = {
-                      next: function (x) {
-                          var destination = this.destination;
-                          destination.onNext(x);
-                      },
-                      error: function (err) {
-                          var destination = this.destination;
-                          if (destination.onError) {
-                              destination.onError(err);
-                          }
-                      },
-                      complete: function () {
-                          var destination = this.destination;
-                          if (destination.onCompleted) {
-                              destination.onCompleted();
-                          }
-                      },
-                      destination: observerOrNextFn
-                  }
-                }
-              subscription = _subscribe.call(this, observer);
-              break;
-          case 'undefined':
-              subscription = _subscribe.call(this);
-              break;
-          default:
-              throw new TypeError('cannot subscribe to observable with ' +
-                  'type ' + typeof observerOrNextFn);
-      }
+    rxNewObservable.subscribe = function(observerOrNextFn, errFn, compFn) {
+        var subscription;
+        if (typeof observerOrNextFn !== "object" || observerOrNextFn === null) {
+            subscription = _subscribe.call(
+                this,
+                observerOrNextFn,
+                errFn,
+                compFn
+            );
+        } else {
+            var observer = toRxNewObserver(observerOrNextFn);
+            subscription = _subscribe.call(this, observer);
+        }
 
-      var _unsubscribe = subscription.unsubscribe;
+        var _unsubscribe = subscription.unsubscribe;
 
-      subscription.unsubscribe = subscription.dispose = function () {
-          this.isDisposed = true;
-          _unsubscribe.call(subscription);
-      };
+        subscription.unsubscribe = subscription.dispose = function() {
+            this.isDisposed = true;
+            _unsubscribe.call(this);
+        };
 
-      return subscription;
-    }
+        return subscription;
+    };
 
     return rxNewObservable;
-}
+};

test/unit/internal/rxNewToRxNewAndOld.spec.js

@@ -27,6 +27,21 @@ describe('rxNewToRxNewAndOld', function () {
         expect(results).to.deep.equal([1, 2, 3, 'done']);
     });
 
+    it('should work with partial "old" observers', function () {
+        var source = Observable.of(1, 2, 3);
+        var results = [];
+
+        var sub = rxNewToRxNewAndOld(source).subscribe({
+            onCompleted: function () {
+                results.push('done');
+            }
+        });
+
+        expect(sub.dispose).to.be.a('function');
+        expect(sub.unsubscribe).to.be.a('function');
+        expect(results).to.deep.equal(['done']);
+    });
+
     it('should work with "new" observers', function () {
         var source = Observable.of(1, 2, 3);
         var results = [];