更智能的缓冲区答案

【问题标题】：Smarter buffers更智能的缓冲区
【发布时间】：2015-11-06 01:59:51
【问题描述】：

我发送请求 - 获取数据数组。为了操纵该数据，我需要将其展平，因此我可以将其用作实体流而不是实体数组流，但是，副作用是我希望这些实体立即出现在 UI 中，而不是一个一个出现，因此它一次只更新一个 UI。

假设我有这样的代码：

// this generates a sequence of objects - getCasesStream sends an ajax request,
// whenever dateRange changes 
casesStm = dateRangeStm.flatMapLatest(getCasesStream)

casesStm.subscribe((x)=> { console.log(x) })

function getCasesStream(dateRange) {
    return getCases(dateRange.startDate, dateRange.endDate)  
        // api every time returns an array,
        // but it's difficult to work with array of elements, ergo the flattening
        .flatMap((x) => x)                                
        .filter((x) => _.isNotEmpty(x.value))
        .map((caseDoc) => _.assign(caseDoc.value, {
            key: caseDoc.id
        }));
}

这很好用，一次发出一个值。现在我想要的是最多发出 10 个值，如果少于 10 个 - 发出剩下的任何值。

我想我可以通过这样做来解决这个问题：

casesStm
    .windowWithCount(10)
    .flatMap((x)=> x.toArray())

但这只有在getCasesStream（过滤后）返回至少 10 个项目时才有效，如果它少于那个 - 我什至不会看到它们。

如何在这里有效地缓冲元素？再次：

api 向我们发送一个数组
要过滤并为每个元素添加额外的道具，最好扁平化该数组（或者不扁平化？）
最后我需要缓冲（不想每次新元素出现时都强制浏览器重绘）

也许我应该使用通用的window，它返回getCasesStream 中元素的长度，但该函数不接受任何参数，我怎样才能得到长度？我尝试使用windowWithTimeOrCount - 即使没有元素，它也会在每个间隔持续发射空缓冲区。

【问题讨论】：

发出 10 个项目是否意味着对于任何一个 getCases 请求，您最多只想显示 10 个案例？

标签： javascript reactive-programming rxjs

【解决方案1】：

您可以过滤掉那些空缓冲区，或者您可以查看How to create a RxJS buffer that groups elements in NodeJS but that does not rely on forever running interval? 中提到的一些选项。

这里展示的想法是将运算符buffer 与关闭选择器一起使用。作为关闭选择器，您可以使用merge(source.skip(9).take(1).repeat(), source.delay(Xms))（或上述链接中建议的运算符flatMapFirst，查看两个选项）。这样，原则上，当没有发出 case 时，没有发出缓冲区，当 case 到达时，merge 运算符在第 10 个 case 或 Xms 之后发出一个值，以先到者为准。当merge 运算符的值被发出时，缓冲区被关闭并发出。

您可以从这里的代码中获得灵感：

function emits(who){
  return function (x) { console.log([who, "emits"].join(" ") + " " + x + " click(s)");};
}

var Xms = 1700;

var source = Rx.Observable.fromEvent(document.body, 'click');
console.log("running");

var delayedSource$ = Rx.Observable.merge(source.skip(9).take(1).repeat(), source.delay(Xms));

var buffered$ = source
     .buffer(function () { return  delayedSource$;}).map(function(clickBuffer){return clickBuffer.length;});

buffered$.subscribe(emits("buffer"));

jsbin : http://jsbin.com/siqopuxoli/edit?html,js,console,output

重要提示：您还应该分享您的案例来源（除非您知道它已经是热门来源），因为它将被多次订阅：getCases(dateRange.startDate, dateRange.endDate).share()

【讨论】：