akarnokd / akarnokd/RxJavaFiberInterop

Cancellation Notification for Long-Running Tasks in the FiberInterop.create

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

描述

When using RxJava 3 in combination with the FiberInterop library, there is a challenge in handling cancellations for long-running tasks. Specifically, the current implementation does not provide a mechanism for a long-running function to be notified of a cancellation until `emit` is invoked. This causes an issue where a function like `computeValue` cannot be aware of a cancellation event, leading to potential inefficiencies and unnecessary computations.

```java
var executor = Executors.newVirtualThreadPerTaskExecutor();
var scheduler = Schedulers.from(executor);
var cancelled = new AtomicBoolean(false);

var flow = FiberInterop.create(emitter -> {
var value = computeValue(cancelled);
emitter.emit(value);
}, executor).timeout(2, TimeUnit.SECONDS, scheduler);

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

In the above code, `computeValue` is a long-running function that takes a `cancelled` flag to potentially halt its execution if the flow is cancelled. However, the `cancelled` flag cannot be updated until the `emit` is called, making it ineffective in stopping the computation early.

To solve this problem, introducing a mechanism like registering `onCancel` callback or setting an interrupt flag on the virtual thread could effectively signal cancellation to the running task.
Here are some examples of how it might look:

```java
var flow = FiberInterop.create(emitter -> {
emitter.onCancel(() -> cancelled.set(true));
var value = computeValue(cancelled);
if (!cancelled.get()) {
emitter.emit(value);
}
}, executor).timeout(2, TimeUnit.SECONDS, scheduler);
```

```java
var flow = FiberInterop.create(emitter -> {
var value = computeValue();
emitter.emit(value);
}, executor, interruptWhenCancelled).timeout(2, TimeUnit.SECONDS, scheduler);
```

This approach ensures that the `computeValue` function can be notified of the cancellation event promptly and can stop its execution accordingly.

贡献指南

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

评估

这个 Issue 还没有评估数据。

把新 issue 发到你的邮箱

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