如何根据发出的事件有条件地缓冲分组 Observable/Flux？

Question

我正在尝试根据以下信息编写反应流：

我们有一个实体事件流，其中每个事件都包含其实体的 ID 和 INTENT 或 COMMIT 类型。假定具有给定 ID 的 COMMIT 之前总是有一个或多个具有相同 ID 的 INTENT。收到 INTENT 时，应按其 ID 对其进行分组，并应打开该组的 "buffer"。当接收到同一组的 COMMIT 或配置的超时已过时，缓冲区应为 "closed"。应发出生成的缓冲区。

请注意，在收到关闭 COMMIT 之前可能会收到多个 INTENT。（编辑：）bufferDuration 应该保证任何 "opened" 缓冲区在收到打开缓冲区的 INTENT 后 bufferDuration 时间后发出，有或没有 COMMIT。

我最近的尝试如下：

public EntityEventBufferFactory {
    private final Duration bufferDuration;

    public EntityEventBufferFactory(Duration bufferDuration) {
        this.bufferDuration = bufferDuration;
    }

    public Flux<List<EntityEvent>> createGroupBufferFlux(Flux<EntityEvent> eventFlux) {
        return eventFlux.groupBy(EntityEvent::getId)
            .map(groupedFlux -> createGroupBuffer(groupedFlux))
            .flatMap(Function.identity());
    }

    protected Flux<List<EntityEvent>> createGroupBuffer(Flux<EntityEvent> groupFlux) {
        return groupFlux.publish().buffer(groupFlux.filter(this::shouldOpenBufferOnEvent), createGroupBufferCloseSelector(groupFlux));
    }

    protected Function<EntityEvent, Publisher<EntityEvent>> createGroupBufferCloseSelector(Flux<EntityEvent> groupFlux) {
        return event -> Flux.firstEmitting(Flux.just(event).delay(bufferDuration), groupFlux.filter(this::shouldCloseBufferOnEvent).publish());
    }

    protected boolean shouldOpenBufferOnEvent(EntityEvent entityEvent) {
        return entityEvent.getEventType() == EventType.INTENT;
    }

    protected boolean shouldCloseBufferOnEvent(EntityEvent entityEvent) {
        return entityEvent.getEventType() == EventType.COMMIT;
    }
}

这是我试图通过的测试：

@Test
public void entityEventsCanBeBuffered() throws Exception {
    FluxProcessor<EntityEvent, EntityEvent> eventQueue = UnicastProcessor.create();

    Duration bufferDuration = Duration.ofMillis(250);

    Flux<List<EntityEvent>> bufferFlux = new EntityEventBufferFactory(bufferDuration).createGroupBufferFlux(eventQueue);
    bufferFactory.setBufferDuration(bufferDuration);

    List<List<EntityEvent>> buffers = new ArrayList<>();
    bufferFlux.subscribe(buffers::add);

    EntityEvent intent = new EntityEvent();
    intent.setId("SOME_ID");
    intent.setEventType(EventType.INTENT);

    EntityEvent commit = new EntityEvent();
    commit.setId(intent.getId());
    commit.setEventType(EventType.COMMIT);

    eventQueue.onNext(intent);
    eventQueue.onNext(commit);

    eventQueue.onNext(intent);
    eventQueue.onNext(commit);

    Thread.sleep(500);

    assertEquals(2, buffers.size());
    assertFalse(buffers.get(0).isEmpty());
    assertFalse(buffers.get(1).isEmpty());
}

通过这个测试，我得到了两个发出的缓冲区，但它们都是空的。您会注意到，在深入研究之后，我不得不在某些点添加 .publish()，以免从 Reactor 获得异常 This processor allows only a single Subscriber。这个问题的答案是我采用这种方法的原因。

我目前正在使用 Reactor，但我认为这与使用 Observable 和同名方法的 RxJava 一对一转换。

有什么想法吗？

Answer 1

我认为这是 Rx 的最终用例 groupBy。来自文档：

Groups the items emitted by a Publisher according to a specified criterion, and emits these grouped items as GroupedFlowables. The emitted GroupedPublisher allows only a single Subscriber during its lifetime and if this Subscriber cancels before the source terminates, the next emission by the source having the same key will trigger a new GroupedPublisher emission.

在您的情况下，此标准是 ID，并且在每个向您发出的 GroupedPublisher 上 takeUntil 类型是 COMMIT：

source
.groupBy(EntityEvent::getId)
.flatMap(group -> 
    group
    .takeUntil(Flowable.timer(10,TimeUnit.SECONDS))
    .takeUntil(this::shouldCloseBufferOnEvent)
    .toList())

编辑：添加时间条件。

Answer 2

感谢 Tassos Bassoukos 的意见。以下 Reactor 代码适用于我：

public EntityEventBufferFactory {
    private final Duration bufferDuration;

    public EntityEventBufferFactory(Duration bufferDuration) {
        this.bufferDuration = bufferDuration;
    }

    @Override
    public Flux<List<EntityEvent>> create(Flux<EntityEvent> eventFlux) {
        return eventFlux.groupBy(EntityEvent::getId)
            .map(this::createGroupBuffer)
            .flatMap(Function.identity());
    }

    protected Mono<List<EntityEvent>> createGroupBuffer(Flux<EntityEvent> groupFlux) {
        return groupFlux.take(bufferDuration)
            .takeUntil(this::shouldCloseBufferOnEvent)
            .collectList();
    }

    protected boolean shouldCloseBufferOnEvent(EntityEvent EntityEvent) {
        return EntityEvent.getEventType() == EventType.COMMIT;
    }
}

如何根据发出的事件有条件地缓冲分组 Observable/Flux？

How do I conditionally buffer a Grouped Observable/Flux based on Emitted Events?

java

reactive-programming

rx-java

project-reactor

rx-java2