javascript - Rxjs 包装 D3 (Observable) 中的其他库函数

Question

我只是不明白这个概念，正在寻找启示。我试图观察何时获取数据，但过程混淆了。这就是我到目前为止所拥有的。 ajax请求是通过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);

虽然这个 ajax 请求在技术上可行，但所有 Observable 函数都在数据到达之前发生。我将如何构造这个，让我的“onNext”日志为我提供数据，而不是仅在 d3.tsv 函数内部获取数据？

Answer 1

有一些 RxJS 运算符专门用于将回调转换为可观察量(.fromCallback、.fromNodeCallback)。然而，它们在这里不起作用，因为它们期望一个回调，而该回调是最后一个参数。这里有两个回调，一个用于成功，一个用于结果。我不知道针对这种情况有任何特殊的运算符，因此我建议您使用自定义辅助函数。

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);