Exploring FromEventPattern in base RxJS with sample implementations in jQuery, DOM, Dojo and Node.js

fromEventPattern.js

Observable.fromEventPattern = function (addHandler, removeHandler) {
	return new AnonymousObservable(function (observer) {
		function innerHandler (e) {
			observer.onNext(e);
		}

		var returnValue = addHandler(innerHandler);
		return disposableCreate(function () {
			if (removeHandler) {
				removeHandler(innerHandler, returnValue);
			}
		});
	}).publish().refCount(); // Ensure only one handler
};

usage-dojo.js

// Using Dojo as an example
require(["dojo/on", "dojo/dom", "dojo/domReady!"], function (on, dom) {
    var target = dojo.byId('target');
    
    var source = Rx.Observable.fromEventPattern(
        function add (h) {
            return on('mousemove', h);        
        },
        function remove (h, signal) {
            // passes signal as second argument
            signal.remove();
        });
        
    var subscription = source.subscribe(function (e) {
        console.log(e.clientX);
    });
});

usage-dom.js

// Plain ol DOM
function onLoad() {
    var target = document.querySelector('#target');

    var source = Rx.Observable.fromEventPattern(
        function add (h) {
            target.addEventListener('mousemove', h, false);      
        },
        function remove (h, signal) {
            target.removeEventListener('mousemove', h, false);   
        });
        
    var subscription = source.subscribe(function (e) {
        console.log(e.clientX);
    });    
}

window.addEventListener('load', onLoaded, false);

usage-jquery.js

// Using jQuery as an example for events
$(function () {
    var target = $('#target');
    
    var source = Rx.Observable.fromEventPattern(
        function add (h) {
            target.bind('mousemove', h);        
        },
        function remove (h) {
            target.unbind('mousemove', h);
        });
        
    var subscription = source.subscribe(function (e) {
        console.log(e.clientX);
    });
});

usage-node.js

// Using Node.js event emitters
var Rx = require('rx');
var EventEmitter = require('events').EventEmitter;

var e = new EventEmitter();

var source = Rx.Observable.fromEventPattern(
    function add (h) {
        target.on('data', h);      
    },
    function remove (h, signal) {
        target.off('data', h);   
    });
    
var subscription = source.subscribe(function (e) {
    console.log(e);
});   

// Emit an event
e.emit('data', 'foo');

// => foo

when does the remove function gets called? in all cases they always add