Rx.trailBy

Methods to convert between Knockout and Reactive Extensions observables

HTML

<link rel="stylesheet" href="http://code.jquery.com/qunit/qunit-1.11.0.css">
<script src="http://code.jquery.com/qunit/qunit-1.11.0.js"></script>
<script src="http://cdn.strategiqcommerce.com/ajax/libs/rxjs/2.1.0/rx.min.js"></script>
<div id="qunit"></div>
<div id="qunit-fixture"></div>

JavaScript

Rx.Observable.prototype.trailBy = function (n) {
    /// <summary>
    /// Converts the source stream into a new stream where the events are delayed by <c>n</c> events.
    /// For example, source.trailBy(10) will yield the first event when <c>source</c> produces the 11th event.
    /// </summary>
    /// <param name="n" type="Number">How many events to lag behind.  0 would produce the same stream as source.  Must not be negative</param>
    /// <returns type="Rx.Observable"></returns>

    var source = this;
    if (n < 0) { throw new Error("n must not be negative"); }
    if (n === 0) { return source; }
    return Rx.Observable.createWithDisposable(function (observer) {
        var ringBuffer = new Array(n),
            i = 0,
            ready;

        return source.subscribe(function (value) {
            if (ready) { observer.onNext(ringBuffer[i]); }
            ringBuffer[i++] = value;
            if (i === n) {
                ready = true;
                i = 0;
            }
        }, observer.onError.bind(observer), observer.onCompleted.bind(observer));
    });
};


QUnit.config.notrycatch = true;
// unit tests
QUnit.module("trailBy()");

QUnit.test("trails by the correct amount", 1, function () {
    var sourceData = [9, 8, 7, 6, 5, 4, 3, 2, 1],
        expected =   [9, 8, 7, 6, 5, 4];
    
    Rx.Observable.fromArray(sourceData).trailBy(3).toArray().subscribe(function (result) {
        QUnit.deepEqual(result, expected, "Correctly trailed by 3");
    });
});