Handle AsyncListener.onComplete in Servlet adapter

Typically the Mono<Void> from the HttpHandler also reflects the
completion of the request and response body processors and at that
point invoking AsyncContext#complete() from HandlerResultSubscriber
should be sufficient.

This commit explicitly propagates the AsyncListener.onComplete event
to the request and response body processors for added safety.
Technically as mentioned those processors should have completed but
depending on how the controller is written there is a possibility
the body processors may not have completed.

Issue: SPR-14772

spring-web/src/main/java/org/springframework/http/server/reactive/ServletHttpHandlerAdapter.java

@@ -91,7 +91,7 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
 				servletRequest, this.dataBufferFactory, this.bufferSize);
 		ServletServerHttpResponse response = new ServletServerHttpResponse(
 				servletResponse, this.dataBufferFactory, this.bufferSize);
-		asyncContext.addListener(new ErrorHandlingAsyncListener(request, response));
+		asyncContext.addListener(new EventHandlingAsyncListener(request, response));
 		HandlerResultSubscriber resultSubscriber = new HandlerResultSubscriber(asyncContext);
 		this.handler.handle(request, response).subscribe(resultSubscriber);
 	}
@@ -131,14 +131,14 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
 	}
 
 
-	private static final class ErrorHandlingAsyncListener implements AsyncListener {
+	private static final class EventHandlingAsyncListener implements AsyncListener {
 
 		private final ServletServerHttpRequest request;
 
 		private final ServletServerHttpResponse response;
 
 
-		public ErrorHandlingAsyncListener(ServletServerHttpRequest request,
+		public EventHandlingAsyncListener(ServletServerHttpRequest request,
 				ServletServerHttpResponse response) {
 
 			this.request = request;
@@ -169,7 +169,8 @@ public class ServletHttpHandlerAdapter extends HttpServlet {
 
 		@Override
 		public void onComplete(AsyncEvent event) {
-			// no op
+			this.request.handleAsyncListenerComplete();
+			this.response.handleAsyncListenerComplete();
 		}
 	}
 

spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpRequest.java

@@ -177,6 +177,13 @@ public class ServletServerHttpRequest extends AbstractServerHttpRequest {
 		}
 	}
 
+	/** Handle a complete callback from the Servlet container */
+	void handleAsyncListenerComplete() {
+		if (this.bodyPublisher != null) {
+			this.bodyPublisher.onAllDataRead();
+		}
+	}
+
 	private RequestBodyPublisher createBodyPublisher() throws IOException {
 		RequestBodyPublisher bodyPublisher = new RequestBodyPublisher(
 				this.request.getInputStream(), this.dataBufferFactory, this.bufferSize);

spring-web/src/main/java/org/springframework/http/server/reactive/ServletServerHttpResponse.java

@@ -166,6 +166,18 @@ public class ServletServerHttpResponse extends AbstractListenerServerHttpRespons
 		}
 	}
 
+	/** Handle a complete callback from the Servlet container */
+	void handleAsyncListenerComplete() {
+		if (this.bodyFlushProcessor != null) {
+			this.bodyFlushProcessor.cancel();
+			this.bodyFlushProcessor.onComplete();
+		}
+		if (this.bodyProcessor != null) {
+			this.bodyProcessor.cancel();
+			this.bodyProcessor.onComplete();
+		}
+	}
+
 
 	private class ResponseBodyProcessor extends AbstractResponseBodyProcessor {