使用 RxJS Observable 流式传输 JSON

Question

我试图了解一些关于 RxJs 的事情。我想做的是使用一些 JSON 数据，并在数据进入时立即开始在 DOM 上呈现该数据。我已经设置了流请求、响应和显示。它的每一个输出都很好，但它是一次性完成的，而不是随着时间的推移。

我想在数据进入时开始在页面上显示数据，而不是等待整个文件完成然后立即显示，这会产生很长的等待时间。

//Cache the selector
var $resultList = $('.results');

//Gets the JSON (this will not be a static file,just for testing)
var requestStream = Rx.Observable.just("/results.json");

var responseStream = requestStream
    .flatMap(function(requestUrl) {
            return Rx.Observable.fromPromise(jQuery.getJSON(requestUrl))
             });

var displayStream = responseStream.subscribe(
    function(response) {
    //This maps to a Handlebars Template and puts it on the DOM
    $resultList.html(compiledTemplate(response)); 
            },
            function(err) {
                    console.log('Error: %s', err);
             },
             function() {
                    console.log('Completed');
             });




//Sample of the data from the JSON file
Object{
    beginIndex: "1"
    catId: "111"
    endIndex: "1"
    products: Array[100]

}

Answer 1

如果我理解得很好，有两点需要说明：

1. 您需要找到一种方法从该文件中获取对象流 当你读完那个文件时，而不是一个大对象 (I want to start showing the data on the page as its coming in)。这 其机制首先取决于源的结构（文件和文件读取机制） 比在 Rxjs 上（每一行都是一个可以导致信息的对象 显示等？）。一旦你有了“最小可显示信息单元”，你就可以在需要时使用 Rxjs 来缓冲/处理它（你想为每个对象或每 100 个对象显示一些东西，还是删除不必要的属性等？）
2. 您需要逐步更新您的显示 随着新数据的到来。这意味着你需要类似的东西 $resultList.html($resultList.html() + compiledTemplate(response)); 将新编译的html 附加到旧的。

更新：为了分块一个数组，你可以看看这个 jsfiddle：http://jsfiddle.net/429vw0za/

var ta_result = document.getElementById('ta_result');

function emits ( who, who_ ) {return function ( x ) {
 who.innerHTML = [who.innerHTML, who_ + " emits " + JSON.stringify(x)].join("\n");
};}

function fillArrayWithNumbers(n) {
        var arr = Array.apply(null, Array(n));
        return arr.map(function (x, i) { return {prop1: i, prop2:i, prop3:i} });
    }

var sampleObj = {
    beginIndex: "1",
    catId: "111",
    endIndex: "1",
    products: fillArrayWithNumbers(100)
}

console.log('sampleObj', sampleObj);

var result$ = Rx.Observable
  .from(sampleObj.products)
  .bufferWithCount(10)
  .map(function(mini_array){return {
  beginIndex: sampleObj.beginIndex,
  catId: sampleObj.catId,
  endIndex: sampleObj.endIndex,
  products: mini_array
  }})
  .do(emits(ta_result, 'result'));

result$.subscribe(function(){    });

然后，您将拥有从大小为 100 的数组中提取的大小为 10 的数组的对象流。