akarnokd / akarnokd/RxJavaFiberInterop

Proposal to Introduce generate Method and Enhance create for Concurrency in the FiberInterop

未关闭
#99 0 条评论 0 个 reaction 已指派 0 人 在 GitHub 查看
主要语言
Java
星标
40
派生
7
PR 合并指标
30 天内没有已合并 PR

描述

The FiberInterop library allows creating an RxJava `Flowable` stream easily, as demonstrated in the following example:

```java
var executor = Executors.newVirtualThreadPerTaskExecutor();

var flow = FiberInterop.create(emitter -> {
for (var i = 0; i < 10; i++) {
Thread.sleep(500);
emitter.emit(i);
}
}, executor);

flow.blockingForEach(value -> {
System.out.println(value);
});
```

This example works perfectly for simple sequential flow generation.
However, issues arise when concurrency is introduced into the flow generation process. For example:

```java
var executor = Executors.newVirtualThreadPerTaskExecutor();
var taskCount = 10;

while (true) {
var flow = FiberInterop.create(emitter -> {
try (var scope = Executors.newVirtualThreadPerTaskExecutor()) {
var latch = new CountDownLatch(1);

for (var i = 0; i < taskCount; i++) {
final var taskId = i;

scope.submit(() -> {
try {
latch.await();
emitter.emit(taskId);
} catch (Throwable e) {
throw new RuntimeException(e);
}
});
}

latch.countDown();
}
}, executor);

var size = flow.toList().blockingGet().size();

if (size < taskCount) {
System.out.println(STR."Expected \{taskCount} but got \{size}");
break;
}
}
```

As can be seen from the example above, the `create` method does not handle concurrency, leading to a situation where the resulting flow size can be less than `taskCount`. This behavior is not documented, which can confuse users who might expect concurrent generation to work seamlessly.

To address this issue, it would be beneficial to differentiate between methods designed for simple non-concurrent workflows and those that handle concurrency by default. Specifically:

1. Introduce a `generate` method. This method would be intended for simple, non-concurrent workflows. It would clearly indicate to users that it is not designed for concurrent flow generation.
2. Enhance the `create` method. Modify the `create` method to handle concurrency properly by default. This would involve ensuring that concurrent tasks are managed and their emissions are correctly handled to prevent missing items.

By introducing the `generate` method and enhancing the `create` method to handle concurrency, users would have the opportunity to select an appropriate approach for handling their tasks.
This would provide functionality similar to the distinction between [flow](https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/flow.html) and [channelFlow](https://kotlinlang.org/api/kotlinx.coroutines/kotlinx-coroutines-core/kotlinx.coroutines.flow/channel-flow.html) in Kotlin.

贡献指南

这个仓库没有索引到贡献指南

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

精选适合新手参与的 GitHub issue 摘要。