momentary_and_cache.js

var Kefir = require("kefir"),
    asap = require("asap");

function cacheStream(source){
    var eventCache = null;
    var sink = Kefir.stream(function(emitter){
        var handler = function(event){
            if(event.type==="value"){
                eventCache = event;
            }
            emitter.emitEvent(event);
        };
        source.onAny(handler);
        return function(){
            source.offAny(handler);
        }
    });
    var oldOnValue = sink.onValue;
    sink.onValue = function(subscriber){
        if(eventCache !== null){
            subscriber(eventCache.value);
        }
        return oldOnValue.apply(this, arguments);
    };
    var oldOnAny = sink.onAny;
    sink.onAny = function(subscriber){
        if(eventCache !== null){
            subscriber(eventCache);
        }
        return oldOnAny.apply(this, arguments);
    };
    return sink;
}

function momentaryStream(source){
    return Kefir.stream(function(emitter){
        var momentaryEvent = null;
        var dispatch = function(){
            emitter.emitEvent(momentaryEvent);
            momentaryEvent = null;
        };
        var handler = function(event){
            if(momentaryEvent === null){
                asap(dispatch);
            }
            momentaryEvent = event;
        };
        source.onAny(handler);
        return function(){
            source.offAny(handler);
        }
    });
}

function onceStreamFactory(){
    return Kefir.stream(function(emitter){
        emitter.emit(1);
        emitter.emit(2);
        emitter.emit(3);
    });
}

var cached = cacheStream(onceStreamFactory());
cached.log("once");
cached.log("twice");


var momentary = momentaryStream(onceStreamFactory());
momentary.log("once");
momentary.log("twice");