Fixing single behavior

Author: mattpodwysocki

dist/rx.aggregates.js

@@ -622,30 +622,76 @@
     }, source);
   };
 
-  /**
-   * Returns the only element of an observable sequence that satisfies the condition in the optional predicate, and reports an exception if there is not exactly one element in the observable sequence.
-   * @param {Function} [predicate] A predicate function to evaluate for elements in the source sequence.
-   * @param {Any} [thisArg] Object to use as `this` when executing the predicate.
-   * @returns {Observable} Sequence containing the single element in the observable sequence that satisfies the condition in the predicate.
-   */
-  observableProto.single = function (predicate, thisArg) {
-    if (isFunction(predicate)) { return this.filter(predicate, thisArg).single(); }
-    var source = this;
-    return new AnonymousObservable(function (o) {
-      var value, seenValue = false;
-      return source.subscribe(function (x) {
-        if (seenValue) {
-          o.onError(new Error('Sequence contains more than one element'));
-        } else {
-          value = x;
-          seenValue = true;
+  var SingleObserver = (function(__super__) {
+    inherits(SingleObserver, __super__);
+    function SingleObserver(o, obj, s) {
+      this._o = o;
+      this._obj = obj;
+      this._s = s;
+      this._i = 0;
+      this._hv = false;
+      this._v = null;
+      __super__.call(this);
+    }
+
+    SingleObserver.prototype.next = function (x) {
+      var shouldYield = false;
+      if (this._obj.predicate) {
+        var res = tryCatch(this._obj.predicate)(x, this._i++, this._s);
+        if (res === errorObj) { return this._o.onError(res.e); }
+        Boolean(res) && (shouldYield = true);
+      } else if (!this._obj.predicate) {
+        shouldYield = true;
+      }
+      if (shouldYield) {
+        if (this._hv) {
+          return this._o.onError(new Error('Sequence contains more than one matching element'));
         }
-      }, function (e) { o.onError(e); }, function () {
-        o.onNext(value);
-        o.onCompleted();
-      });
-    }, source);
-  };
+        this._hv = true;
+        this._v = x;
+      }
+    };
+    SingleObserver.prototype.error = function (e) { this._o.onError(e); };
+    SingleObserver.prototype.completed = function () {
+      if (this._hv) {
+        this._o.onNext(this._v);
+        this._o.onCompleted();
+      }
+      else if (this._obj.defaultValue === undefined) {
+        this._o.onError(new EmptyError());
+      } else {
+        this._o.onNext(this._obj.defaultValue);
+        this._o.onCompleted();
+      }
+    };
+
+    return SingleObserver;
+  }(AbstractObserver));
+
+
+    /**
+     * Returns the only element of an observable sequence that satisfies the condition in the optional predicate, and reports an exception if there is not exactly one element in the observable sequence.
+     * @returns {Observable} Sequence containing the single element in the observable sequence that satisfies the condition in the predicate.
+     */
+    observableProto.single = function (predicate, thisArg) {
+      var obj = {}, source = this;
+      if (typeof arguments[0] === 'object') {
+        obj = arguments[0];
+      } else {
+        obj = {
+          predicate: arguments[0],
+          thisArg: arguments[1],
+          defaultValue: arguments[2]
+        };
+      }
+      if (isFunction (obj.predicate)) {
+        var fn = obj.predicate;
+        obj.predicate = bindCallback(fn, obj.thisArg, 3);
+      }
+      return new AnonymousObservable(function (o) {
+        return source.subscribe(new SingleObserver(o, obj, source));
+      }, source);
+    };
 
   var FirstObserver = (function(__super__) {
     inherits(FirstObserver, __super__);
@@ -707,6 +753,49 @@
     }, source);
   };
 
+  var LastObserver = (function(__super__) {
+    inherits(LastObserver, __super__);
+    function LastObserver(o, obj, s) {
+      this._o = o;
+      this._obj = obj;
+      this._s = s;
+      this._i = 0;
+      this._hv = false;
+      this._v = null;
+      __super__.call(this);
+    }
+
+    LastObserver.prototype.next = function (x) {
+      var shouldYield = false;
+      if (this._obj.predicate) {
+        var res = tryCatch(this._obj.predicate)(x, this._i++, this._s);
+        if (res === errorObj) { return this._o.onError(res.e); }
+        Boolean(res) && (shouldYield = true);
+      } else if (!this._obj.predicate) {
+        shouldYield = true;
+      }
+      if (shouldYield) {
+        this._hv = true;
+        this._v = x;
+      }
+    };
+    LastObserver.prototype.error = function (e) { this._o.onError(e); };
+    LastObserver.prototype.completed = function () {
+      if (this._hv) {
+        this._o.onNext(this._v);
+        this._o.onCompleted();
+      }
+      else if (this._obj.defaultValue === undefined) {
+        this._o.onError(new EmptyError());
+      } else {
+        this._o.onNext(this._obj.defaultValue);
+        this._o.onCompleted();
+      }
+    };
+
+    return LastObserver;
+  }(AbstractObserver));
+
   /**
    * Returns the last element of an observable sequence that satisfies the condition in the predicate if specified, else the last element.
    * @returns {Observable} Sequence containing the last element in the observable sequence that satisfies the condition in the predicate.
@@ -727,34 +816,7 @@
       obj.predicate = bindCallback(fn, obj.thisArg, 3);
     }
     return new AnonymousObservable(function (o) {
-      var value, seenValue = false, i = 0;
-      return source.subscribe(
-        function (x) {
-          if (obj.predicate) {
-            var res = tryCatch(obj.predicate)(x, i++, source);
-            if (res === errorObj) { return o.onError(res.e); }
-            if (res) {
-              seenValue = true;
-              value = x;
-            }
-          } else if (!obj.predicate) {
-            seenValue = true;
-            value = x;
-          }
-        },
-        function (e) { o.onError(e); },
-        function () {
-          if (seenValue) {
-            o.onNext(value);
-            o.onCompleted();
-          }
-          else if (obj.defaultValue === undefined) {
-            o.onError(new EmptyError());
-          } else {
-            o.onNext(obj.defaultValue);
-            o.onCompleted();
-          }
-        });
+      return source.subscribe(new LastObserver(o, obj, source));
     }, source);
   };