更智能的缓冲区 [英] Smarter buffers

问题描述

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

I send request - get an array of data. In order to manipulate that data I need to flatten it, so I can use it as a stream of entities not stream of array of entities but, then on side-effect I want those entities appear in the UI at once, not one by one, so it updates UI only ones at a time.

假设我有这样的代码:

// 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 个 - 发出剩下的任何值.

This works just fine, emits one value at a time. Now what I want is to emit at most 10 values, if there are less than 10 - emit whatever left.

我以为我可以通过这样做来解决这个问题:

I thought I could solve that by doing this:

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

但是这仅在 getCasesStream(过滤后)返回至少 10 个项目时才有效，如果少于这个数量 - 我什至看不到它们.

But then this only works if getCasesStream (after filtering) returns at least 10 items, if it has less than that - I won't even see them.

如何在这里有效地缓冲元素?再次:

How can I buffer elements effectively here? again:

• api 向我们发送一个数组
• 要过滤并为每个元素添加额外的道具，最好将该数组展平(或者可能不展平?)
• 最后我需要缓冲(不想每次新元素出现时都强制浏览器重绘)

也许我应该使用通用的 window 返回 getCasesStream 中元素的长度，但该函数不带任何参数，我如何获得长度?我尝试使用 windowWithTimeOrCount - 即使没有元素，它也会在每个时间间隔发出空缓冲区.

Maybe I should use generic window that returns length of elements in getCasesStream but that function doesn't take any arguments, how can I get the length? I tried using windowWithTimeOrCount - that keeps emitting empty buffer every interval, even when there's no elements.

推荐答案

您可以过滤掉那些空缓冲区，或者您可以查看 如何创建一个 RxJS 缓冲区来对 NodeJS 中的元素进行分组但没有依靠永远运行间隔?.

You could filter out those empty buffer and alternatively you could review some of the options mentioned in 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 操作符的值被发出时，缓冲区被关闭并发出.

The idea showcased there is to use the operator buffer with closing selector. As a closing selector, you could use merge(source.skip(9).take(1).repeat(), source.delay(Xms)) (or the operator flatMapFirst as proposed as well in the link thereaforementioned, review both options). In that way, in principle when there is no case emitted, there is no buffer emitted, and when a case arrive, the merge operator emits a value either on the 10th case, or Xms after, whichever is first. When the value from the merge operator is emitted, the buffer is closed and emitted.

您可以从这里的代码中获取灵感:

You can take inspiration from the code here:

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,控制台,输出

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

Important note : you should also share your case source (unless you know it is a hot source already), as it will be subscribed several times: getCases(dateRange.startDate, dateRange.endDate).share()

这篇关于更智能的缓冲区的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！