@MessageExceptionHandler supports error signal

Before this change if a controller method returned a Publisher whose
first signal was an error, the error signal would not be delegated to
a @MessageExceptionHandler method as expected.

To make this work for now we use a package private local copy of the
ChannelSendOperator from spring-web.

See gh-21987

Author: rstoyanchev

spring-messaging/src/main/java/org/springframework/messaging/handler/invocation/reactive/AbstractEncoderMethodReturnValueHandler.java

@@ -112,7 +112,8 @@ public Mono<Void> handleReturnValue(
 		Flux<DataBuffer> encodedContent = encodeContent(
 				returnValue, returnType, bufferFactory, mimeType, Collections.emptyMap());
 
-		return handleEncodedContent(encodedContent, returnType, message);
+		return new ChannelSendOperator<>(encodedContent, publisher ->
+				handleEncodedContent(Flux.from(publisher), returnType, message));
 	}
 
 	@SuppressWarnings("unchecked")