英文:
Java reactor how to properly start async cancellable sideeffect
问题
我正在尝试使用 Reactor 编写一些代码，我知道如何使用 CompletableFuture 编写相应的代码。我在其中收到了“在非阻塞范围内调用订阅”的警告。
我的目标是使用超时调用 `turnOn()`，在超时后调用 `turnOff()`。如果再次调用 `turnOn()`，它应该取消旧的超时并等待新的超时。
我应该如何做到这一点？我可以使用 CompletableFuture 进行超时等待，但是 Reactor 的 API 更加简便。
这个测试按预期工作：
```java
public class TimeoutTest {
Service service;
@BeforeEach
public void setUp() {
service = mock(Service.class);
}
CompletableFuture<Void> turnOffFuture = null;
@DisplayName("Should timeout on turnOn with timeout")
@Test
public void timeoutCompletableFuture() throws InterruptedException {
turnOn(Duration.ofMillis(100)).join();
verify(service).turnOn();
verify(service,never()).turnOff();
Thread.sleep(1000);
verify(service).turnOff();
}
private interface Service{
void turnOn();
void turnOff();
}
public void cancelTimeout() {
if (turnOffFuture != null)
turnOffFuture.cancel(false);
turnOffFuture = null;
}
public CompletableFuture<Void> turnOn(Duration timeout) {
CompletableFuture<Void> turnOnFuture = turnOn();
cancelTimeout();
turnOffFuture = turnOnFuture.thenRun(() -> delay(timeout))
.thenRun(this::turnOff);
return turnOnFuture;
}
private void delay(Duration duration) {
try {
Thread.sleep(BigDecimal.valueOf(duration.getSeconds())
.scaleByPowerOfTen(3)
.add(BigDecimal.valueOf(duration.getNano(), 6))
.intValue());
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
private CompletableFuture<Void> turnOn() {
return CompletableFuture.runAsync(() -> service.turnOn());
}
private CompletableFuture<Void> turnOff() {
return CompletableFuture.runAsync(() -> service.turnOff());
}
}
但是我的 Reactor 代码没有按预期工作。
public class TimeoutMonoTest {
Service service;
@BeforeEach
public void setUp() {
service = mock(Service.class);
}
Disposable turnOffDisposable = null;
@DisplayName("Should timeout on turnOn with timeout")
@Test
public void timeoutMono() throws InterruptedException {
turnOn(Duration.ofMillis(100)).block(Duration.ofMillis(10));
verify(service).turnOn();
verify(service, never()).turnOff();
Thread.sleep(1000);
verify(service).turnOff();
}
private interface Service {
void turnOn();
void turnOff();
}
public void cancelTimeout() {
if (turnOffDisposable != null)
turnOffDisposable.dispose();
turnOffDisposable = null;
}
public Mono<Void> turnOn(Duration timeout) {
Mono<Void> turnOnFuture = turnOn();
cancelTimeout();
turnOffDisposable = turnOnFuture.delayElement(timeout)
.subscribe(it -> this.turnOff());
return turnOnFuture;
}
private Mono<Void> turnOn() {
service.turnOn();
return Mono.just("not empty but mapped to void").then();
}
private Mono<Void> turnOff() {
service.turnOff();
return Mono.just("not empty but mapped to void").then();
}
}
英文:
I'm trying to write something using reactor which I know how to write using completable futures. I'm getting "Calling subscribe in non-blocking scope" warning in it.
My goal is to call turnOn() with a timeout which should call turnOff() after the timeout. If turnOn() is called again it should cancel the old timeout and wait for a new timeout.
How should I do this? I could do a hibrate and use CompletableFuture for the timeout but reactor's api is just a bit easier.
this test works as expected:
public class TimeoutTest {
Service service;
@BeforeEach
public void setUp() {
service = mock(Service.class);
}
CompletableFuture<Void> turnOffFuture = null;
@DisplayName("Should timeout on turnOn with timeout")
@Test
public void timeoutCompletableFuture() throws InterruptedException {
turnOn(Duration.ofMillis(100)).join();
verify(service).turnOn();
verify(service,never()).turnOff();
Thread.sleep(1000);
verify(service).turnOff();
}
private interface Service{
void turnOn();
void turnOff();
}
public void cancelTimeout() {
if (turnOffFuture != null)
turnOffFuture.cancel(false);
turnOffFuture = null;
}
public CompletableFuture<Void> turnOn(Duration timeout) {
CompletableFuture<Void> turnOnFuture = turnOn();
cancelTimeout();
turnOffFuture = turnOnFuture.thenRun(() -> delay(timeout))
.thenRun(this::turnOff);
return turnOnFuture;
}
private void delay(Duration duration) {
try {
Thread.sleep(BigDecimal.valueOf(duration.getSeconds())
.scaleByPowerOfTen(3)
.add(BigDecimal.valueOf(duration.getNano(), 6))
.intValue());
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
}
private CompletableFuture<Void> turnOn() {
return CompletableFuture.runAsync(() -> service.turnOn());
}
private CompletableFuture<Void> turnOff() {
return CompletableFuture.runAsync(() -> service.turnOff());
}
}
but my reactor code does not.
public class TimeoutMonoTest {
Service service;
@BeforeEach
public void setUp() {
service = mock(Service.class);
}
Disposable turnOffDisposable = null;
@DisplayName("Should timeout on turnOn with timeout")
@Test
public void timeoutMono() throws InterruptedException {
turnOn(Duration.ofMillis(100)).block(Duration.ofMillis(10));
verify(service).turnOn();
verify(service, never()).turnOff();
Thread.sleep(1000);
verify(service).turnOff();
}
private interface Service {
void turnOn();
void turnOff();
}
public void cancelTimeout() {
if (turnOffDisposable != null)
turnOffDisposable.dispose();
turnOffDisposable = null;
}
public Mono<Void> turnOn(Duration timeout) {
Mono<Void> turnOnFuture = turnOn();
cancelTimeout();
turnOffDisposable = turnOnFuture.delayElement(timeout)
.subscribe(it -> this.turnOff());
return turnOnFuture;
}
private Mono<Void> turnOn() {
service.turnOn();
return Mono.just("not empty but mapped to void").then();
}
private Mono<Void> turnOff() {
service.turnOff();
return Mono.just("not empty but mapped to void").then();
}
}
答案1
得分: 0
问题出在turnOn()和turnOff()方法中对void mono的映射上。实际上，它们并没有获得“下一个”信号，而只是一个“成功”信号。
修复方法很简单，只需要将turnOn方法更改为：
public Mono<Void> turnOn(Duration timeout) {
cancelTimeout();
Mono<Void> turnOnMono = turnOn();
turnOffDisposable = turnOnMono.delayElement(timeout)
.then(turnOff())
.subscribe();
return turnOn();
}
英文:
The problem lies in the mapping to void mono's in the turnOn() and turnOff() methods. They do not actually get a "next" signal, just a "success" signal.
The fix is simply to change the turnOn method to:
public Mono<Void> turnOn(Duration timeout) {
cancelTimeout();
Mono<Void> turnOnMono = turnOn();
turnOffDisposable = turnOnMono.delayElement(timeout)
.then(turnOff())
.subscribe();
return turnOn();
}
通过集体智慧和协作来改善编程学习和解决问题的方式。致力于成为全球开发者共同参与的知识库，让每个人都能够通过互相帮助和分享经验来进步。
评论