Reactive-Extensions/RxJS

View on GitHub
src/core/linq/observable/windowwithcount.js

Summary

Maintainability
A
1 hr
Test Coverage
/**
* Projects each element of an observable sequence into zero or more windows which are produced based on element count information.
* @param {Number} count Length of each window.
* @param {Number} [skip] Number of elements to skip between creation of consecutive windows. If not specified, defaults to the count.
* @returns {Observable} An observable sequence of windows.
*/
Function `windowCount` has 37 lines of code (exceeds 25 allowed). Consider refactoring.
Function `windowCount` has a Cognitive Complexity of 6 (exceeds 5 allowed). Consider refactoring.
observableProto.windowWithCount = observableProto.windowCount = function (count, skip) {
var source = this;
+count || (count = 0);
Math.abs(count) === Infinity && (count = 0);
if (count <= 0) { throw new ArgumentOutOfRangeError(); }
skip == null && (skip = count);
+skip || (skip = 0);
Math.abs(skip) === Infinity && (skip = 0);
 
if (skip <= 0) { throw new ArgumentOutOfRangeError(); }
return new AnonymousObservable(function (observer) {
var m = new SingleAssignmentDisposable(),
refCountDisposable = new RefCountDisposable(m),
n = 0,
q = [];
 
function createWindow () {
var s = new Subject();
q.push(s);
observer.onNext(addRef(s, refCountDisposable));
}
 
createWindow();
 
m.setDisposable(source.subscribe(
function (x) {
for (var i = 0, len = q.length; i < len; i++) { q[i].onNext(x); }
var c = n - count + 1;
c >= 0 && c % skip === 0 && q.shift().onCompleted();
++n % skip === 0 && createWindow();
},
function (e) {
while (q.length > 0) { q.shift().onError(e); }
observer.onError(e);
},
function () {
while (q.length > 0) { q.shift().onCompleted(); }
observer.onCompleted();
}
));
return refCountDisposable;
}, source);
};