提交 09da10cc 编写于 作者: V Violeta Georgieva 提交者: Rossen Stoyanchev

Propagate the cancel signal to the downstream

Issue: SPR-17609
上级 44f39d4c
......@@ -338,6 +338,9 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
private final WriteBarrier writeBarrier;
@Nullable
private Subscription subscription;
public WriteCompletionBarrier(CoreSubscriber<? super Void> subscriber, WriteBarrier writeBarrier) {
this.completionSubscriber = subscriber;
......@@ -357,6 +360,7 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
@Override
public void onSubscribe(Subscription subscription) {
this.subscription = subscription;
subscription.request(Long.MAX_VALUE);
}
......@@ -388,6 +392,10 @@ public class ChannelSendOperator<T> extends Mono<Void> implements Scannable {
@Override
public void cancel() {
this.writeBarrier.cancel();
Subscription subscription = this.subscription;
if (subscription != null) {
subscription.cancel();
}
}
}
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册