From 1c6bab23ea0c5d8fa90b594f962d703575da93dd Mon Sep 17 00:00:00 2001 From: Rossen Stoyanchev Date: Thu, 11 Feb 2021 21:26:37 +0000 Subject: [PATCH] Polishing See gh-26434 --- .../AbstractListenerServerHttpResponse.java | 26 +++++++++---------- .../AbstractListenerWriteFlushProcessor.java | 6 ++--- .../AbstractListenerWriteProcessor.java | 10 +++---- .../reactive/ServletHttpHandlerAdapter.java | 17 +++++++++--- 4 files changed, 34 insertions(+), 25 deletions(-) diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java index 0eed513e27..75c9bc7113 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerServerHttpResponse.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2018 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -38,12 +38,12 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH private final AtomicBoolean writeCalled = new AtomicBoolean(); - public AbstractListenerServerHttpResponse(DataBufferFactory dataBufferFactory) { - super(dataBufferFactory); + public AbstractListenerServerHttpResponse(DataBufferFactory bufferFactory) { + super(bufferFactory); } - public AbstractListenerServerHttpResponse(DataBufferFactory dataBufferFactory, HttpHeaders headers) { - super(dataBufferFactory, headers); + public AbstractListenerServerHttpResponse(DataBufferFactory bufferFactory, HttpHeaders headers) { + super(bufferFactory, headers); } @@ -56,15 +56,15 @@ public abstract class AbstractListenerServerHttpResponse extends AbstractServerH protected final Mono writeAndFlushWithInternal( Publisher> body) { - if (this.writeCalled.compareAndSet(false, true)) { - Processor, Void> processor = createBodyFlushProcessor(); - return Mono.from(subscriber -> { - body.subscribe(processor); - processor.subscribe(subscriber); - }); + if (!this.writeCalled.compareAndSet(false, true)) { + return Mono.error(new IllegalStateException( + "writeWith() or writeAndFlushWith() has already been called")); } - return Mono.error(new IllegalStateException( - "writeWith() or writeAndFlushWith() has already been called")); + Processor, Void> processor = createBodyFlushProcessor(); + return Mono.from(subscriber -> { + body.subscribe(processor); + processor.subscribe(subscriber); + }); } /** diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java index ab0b854124..f8575ab315 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteFlushProcessor.java @@ -58,7 +58,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo @Nullable private Subscription subscription; - private volatile boolean subscriberCompleted; + private volatile boolean sourceCompleted; private final WriteResultPublisher resultPublisher; @@ -293,7 +293,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo return; } if (processor.changeState(this, REQUESTED)) { - if (processor.subscriberCompleted) { + if (processor.sourceCompleted) { handleSubscriberCompleted(processor); } else { @@ -304,7 +304,7 @@ public abstract class AbstractListenerWriteFlushProcessor implements Processo } @Override public void onComplete(AbstractListenerWriteFlushProcessor processor) { - processor.subscriberCompleted = true; + processor.sourceCompleted = true; // A competing write might have completed very quickly if (processor.state.get().equals(State.REQUESTED)) { handleSubscriberCompleted(processor); diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java index 9e026f047d..19c478554f 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/AbstractListenerWriteProcessor.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -64,7 +64,7 @@ public abstract class AbstractListenerWriteProcessor implements Processor implements Processor implements Processor void onComplete(AbstractListenerWriteProcessor processor) { - processor.subscriberCompleted = true; + processor.sourceCompleted = true; // A competing write might have completed very quickly if (processor.state.get().equals(State.REQUESTED)) { processor.changeStateToComplete(State.REQUESTED); @@ -408,7 +408,7 @@ public abstract class AbstractListenerWriteProcessor implements Processor void onComplete(AbstractListenerWriteProcessor processor) { - processor.subscriberCompleted = true; + processor.sourceCompleted = true; // A competing write might have completed very quickly if (processor.state.get().equals(State.REQUESTED)) { processor.changeStateToComplete(State.REQUESTED); diff --git a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java index 750b4fc7d9..54961a8449 100644 --- a/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java +++ b/spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java @@ -1,5 +1,5 @@ /* - * Copyright 2002-2020 the original author or authors. + * Copyright 2002-2021 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -239,15 +239,24 @@ public class ServletHttpHandlerAdapter implements Servlet { } + /** + * AsyncListener to complete the {@link AsyncContext} in case of error or + * timeout notifications from the container + *

Additional {@link AsyncListener}s are registered in + * {@link ServletServerHttpRequest} to signal onError/onComplete to the + * request body Subscriber, and in {@link ServletServerHttpResponse} to + * cancel the write Publisher and signal onError/onComplete downstream to + * the writing result Subscriber. + */ private static class HandlerResultAsyncListener implements AsyncListener { private final AtomicBoolean isCompleted; private final String logPrefix; - public HandlerResultAsyncListener(AtomicBoolean isCompleted, ServletServerHttpRequest httpRequest) { + public HandlerResultAsyncListener(AtomicBoolean isCompleted, ServletServerHttpRequest request) { this.isCompleted = isCompleted; - this.logPrefix = httpRequest.getLogPrefix(); + this.logPrefix = request.getLogPrefix(); } @Override @@ -277,7 +286,7 @@ public class ServletHttpHandlerAdapter implements Servlet { } - private class HandlerResultSubscriber implements Subscriber { + private static class HandlerResultSubscriber implements Subscriber { private final AsyncContext asyncContext;