javascript - Rxjs wrapping other libraries functions from D3 (Observable) -

i not getting concept , looking enlightenment. trying observe when data fetched have got process mixed up. have far. ajax request done d3.tsv.

var test = rx.observable.just(     d3.tsv("https://gist.githubusercontent.com/mbostock/3885304/raw/37bd91278846c053188a130a01770cddff023590/data.tsv",          function(d) {           return {             letter: d.letter,             frequency: +d.frequency           };         },          function(error, rows) {           console.log('mytest2',rows);         }     ) );  var observer = rx.observer.create(   function (x) { console.log('onnext: %s', x); },   function (e) { console.log('onerror: %s', e); },   function () { console.log('oncompleted'); });  var subscription = test.subscribe(observer); 

while ajax request technically works, of observable functions happen before data arrives. how construct 'onnext' log gives me data instead of getting inside of d3.tsv function?

there rxjs operators dedicated convert callbacks observables (.fromcallback, .fromnodecallback). not work here, expect 1 callback, , callback last parameter. here have 2 callbacks, 1 success, 1 result. not aware of special operator case, recommend use custom helper function.

function d3fn (url, success_handler, error_handler) {   success_handler ({     letter : 'letter',     frequency : 9   }); }  var d3 = {tsv : d3fn};  function fromd3callback (d3fn, ctx) { return function () {   var args = array.prototype.slice.call(arguments);   var subject = new rx.asyncsubject();    function success_handler () {     subject.onnext.apply(subject, array.prototype.slice.call(arguments));     subject.oncompleted();   }    function error_handler () {     subject.onerror(array.prototype.slice.call(arguments));   }    args.push(success_handler);   args.push(error_handler);    d3fn.apply(ctx, args);   return subject.asobservable(); } }  var test = fromd3callback(d3.tsv)("https://gist.githubusercontent.com/mbostock/3885304/raw/37bd91278846c053188a130a01770cddff023590/data.tsv")   .map(function(d) {           return {             letter: d.letter,             frequency: +d.frequency           };         })   .catch(function(error, rows) {           console.log('mytest2',rows);           return rx.observable.throw({error: error, rows: rows});         });  var observer = rx.observer.create(   function (x) { console.log('onnext: %o', x); },   function (e) { console.log('onerror: %s', e); },   function () { console.log('oncompleted'); });  var subscription = test.subscribe(observer);