Merge branch '6.1.x'

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandler.java

@@ -337,6 +337,7 @@ class ReactiveTypeHandler {
 						logger.trace("Send for " + this.emitter + " failed: " + ex);
 					}
 					terminate();
+					this.emitter.completeWithError(ex);
 					return;
 				}
 			}

spring-webmvc/src/main/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitter.java

@@ -79,16 +79,6 @@ public class ResponseBodyEmitter {
 	@Nullable
 	private Throwable failure;
 
-	/**
-	 * After an I/O error, we don't call {@link #completeWithError} directly but
-	 * wait for the Servlet container to call us via {@code AsyncListener#onError}
-	 * on a container thread at which point we call completeWithError.
-	 * This flag is used to ignore further calls to complete or completeWithError
-	 * that may come for example from an application try-catch block on the
-	 * thread of the I/O error.
-	 */
-	private boolean ioErrorOnSend;
-
 	private final DefaultCallback timeoutCallback = new DefaultCallback();
 
 	private final ErrorCallback errorCallback = new ErrorCallback();
@@ -198,7 +188,6 @@ public class ResponseBodyEmitter {
 				this.handler.send(object, mediaType);
 			}
 			catch (IOException ex) {
-				this.ioErrorOnSend = true;
 				throw ex;
 			}
 			catch (Throwable ex) {
@@ -234,7 +223,6 @@ public class ResponseBodyEmitter {
 				this.handler.send(items);
 			}
 			catch (IOException ex) {
-				this.ioErrorOnSend = true;
 				throw ex;
 			}
 			catch (Throwable ex) {
@@ -255,10 +243,6 @@ public class ResponseBodyEmitter {
 	 * related events such as an error while {@link #send(Object) sending}.
 	 */
 	public synchronized void complete() {
-		// Ignore complete after IO failure on send
-		if (this.ioErrorOnSend) {
-			return;
-		}
 		this.complete = true;
 		if (this.handler != null) {
 			this.handler.complete();
@@ -277,10 +261,6 @@ public class ResponseBodyEmitter {
 	 * {@link #send(Object) sending}.
 	 */
 	public synchronized void completeWithError(Throwable ex) {
-		// Ignore complete after IO failure on send
-		if (this.ioErrorOnSend) {
-			return;
-		}
 		this.complete = true;
 		this.failure = ex;
 		if (this.handler != null) {

spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ReactiveTypeHandlerTests.java

@@ -16,6 +16,7 @@
 
 package org.springframework.web.servlet.mvc.method.annotation;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -372,6 +373,24 @@ class ReactiveTypeHandlerTests {
 		assertThat(emitterHandler.getValuesAsText()).isEqualTo("The quick brown fox jumps over the lazy dog");
 	}
 
+	@Test
+	void failOnWriteShouldCompleteEmitter() throws Exception {
+
+		Sinks.Many<String> sink = Sinks.many().unicast().onBackpressureBuffer();
+		ResponseBodyEmitter emitter = handleValue(sink.asFlux(), Flux.class, forClass(String.class));
+
+		ErroringEmitterHandler emitterHandler = new ErroringEmitterHandler();
+		emitter.initialize(emitterHandler);
+
+		sink.tryEmitNext("The quick");
+		sink.tryEmitNext(" brown fox jumps over ");
+		sink.tryEmitNext("the lazy dog");
+		sink.tryEmitComplete();
+
+		assertThat(emitterHandler.getHandlingStatus()).isEqualTo(HandlingStatus.ERROR);
+		assertThat(emitterHandler.getFailure()).isInstanceOf(IOException.class);
+	}
+
 	@Test
 	void writeFluxOfString() throws Exception {
 
@@ -451,6 +470,10 @@ class ReactiveTypeHandlerTests {
 
 		private final List<Object> values = new ArrayList<>();
 
+		private HandlingStatus handlingStatus;
+
+		private Throwable failure;
+
 
 		public List<?> getValues() {
 			return this.values;
@@ -460,22 +483,33 @@ class ReactiveTypeHandlerTests {
 			return this.values.stream().map(Object::toString).collect(Collectors.joining());
 		}
 
+		public HandlingStatus getHandlingStatus() {
+			return this.handlingStatus;
+		}
+
+		public Throwable getFailure() {
+			return this.failure;
+		}
+
 		@Override
-		public void send(Object data, MediaType mediaType) {
+		public void send(Object data, MediaType mediaType) throws IOException {
 			this.values.add(data);
 		}
 
 		@Override
-		public void send(Set<ResponseBodyEmitter.DataWithMediaType> items) {
+		public void send(Set<ResponseBodyEmitter.DataWithMediaType> items) throws IOException {
 			items.forEach(item -> this.values.add(item.getData()));
 		}
 
 		@Override
 		public void complete() {
+			this.handlingStatus = HandlingStatus.SUCCESS;
 		}
 
 		@Override
 		public void completeWithError(Throwable failure) {
+			this.handlingStatus = HandlingStatus.ERROR;
+			this.failure = failure;
 		}
 
 		@Override
@@ -491,6 +525,22 @@ class ReactiveTypeHandlerTests {
 		}
 	}
 
+	private enum HandlingStatus {
+		SUCCESS,ERROR
+	}
+
+	private static class ErroringEmitterHandler extends EmitterHandler {
+		@Override
+		public void send(Object data, MediaType mediaType) throws IOException {
+			throw new IOException();
+		}
+
+		@Override
+		public void send(Set<ResponseBodyEmitter.DataWithMediaType> items) throws IOException {
+			throw new IOException();
+		}
+	}
+
 	private static class Bar {
 
 		private final String value;

spring-webmvc/src/test/java/org/springframework/web/servlet/mvc/method/annotation/ResponseBodyEmitterTests.java

@@ -148,23 +148,6 @@ public class ResponseBodyEmitterTests {
 		verifyNoMoreInteractions(this.handler);
 	}
 
-	@Test // gh-30687
-	void completeIgnoredAfterIOException() throws Exception {
-		this.emitter.initialize(this.handler);
-		verify(this.handler).onTimeout(any());
-		verify(this.handler).onError(any());
-		verify(this.handler).onCompletion(any());
-		verifyNoMoreInteractions(this.handler);
-
-		willThrow(new IOException()).given(this.handler).send("foo", MediaType.TEXT_PLAIN);
-		assertThatIOException().isThrownBy(() -> this.emitter.send("foo", MediaType.TEXT_PLAIN));
-		verify(this.handler).send("foo", MediaType.TEXT_PLAIN);
-		verifyNoMoreInteractions(this.handler);
-
-		this.emitter.complete();
-		verifyNoMoreInteractions(this.handler);
-	}
-
 	@Test // gh-30687
 	void completeAfterNonIOException() throws Exception {
 		this.emitter.initialize(this.handler);