/** * Groups the elements of an observable sequence according to a specified key selector function. * A duration selector function is used to control the lifetime of groups. When a group expires, it receives an OnCompleted notification. When a new element with the same * key value as a reclaimed group occurs, the group will be reborn with a new lifetime request. * * @example * var res = observable.groupByUntil(function (x) { return x.id; }, null, function () { return Rx.Observable.never(); }); * 2 - observable.groupBy(function (x) { return x.id; }), function (x) { return x.name; }, function () { return Rx.Observable.never(); }); * 3 - observable.groupBy(function (x) { return x.id; }), function (x) { return x.name; }, function () { return Rx.Observable.never(); }, function (x) { return x.toString(); }); * @param {Function} keySelector A function to extract the key for each element. * @param {Function} durationSelector A function to signal the expiration of a group. * @param {Function} [comparer] Used to compare objects. When not specified, the default comparer is used. * @returns {Observable} * A sequence of observable groups, each of which corresponds to a unique key value, containing all elements that share that same key value. * If a group's lifetime expires, a new group with the same key value can be created once an element with such a key value is encoutered. * */ observableProto.groupByUntil = function (keySelector, elementSelector, durationSelector, comparer) { var source = this; elementSelector || (elementSelector = identity); comparer || (comparer = defaultComparer); return new AnonymousObservable(function (observer) { function handleError(e) { return function (item) { item.onError(e); }; } var map = new Dictionary(0, comparer), groupDisposable = new CompositeDisposable(), refCountDisposable = new RefCountDisposable(groupDisposable); groupDisposable.add(source.subscribe(function (x) { var key; try { key = keySelector(x); } catch (e) { map.getValues().forEach(handleError(e)); observer.onError(e); return; } var fireNewMapEntry = false, writer = map.tryGetValue(key); if (!writer) { writer = new Subject(); map.set(key, writer); fireNewMapEntry = true; } if (fireNewMapEntry) { var group = new GroupedObservable(key, writer, refCountDisposable), durationGroup = new GroupedObservable(key, writer); try { duration = durationSelector(durationGroup); } catch (e) { map.getValues().forEach(handleError(e)); observer.onError(e); return; } observer.onNext(group); var md = new SingleAssignmentDisposable(); groupDisposable.add(md); var expire = function () { map.remove(key) && writer.onCompleted(); groupDisposable.remove(md); }; md.setDisposable(duration.take(1).subscribe( noop, function (exn) { map.getValues().forEach(handleError(exn)); observer.onError(exn); }, expire) ); } var element; try { element = elementSelector(x); } catch (e) { map.getValues().forEach(handleError(e)); observer.onError(e); return; } writer.onNext(element); }, function (ex) { map.getValues().forEach(handleError(ex)); observer.onError(ex); }, function () { map.getValues().forEach(function (item) { item.onCompleted(); }); observer.onCompleted(); })); return refCountDisposable; }, source); };