Fix HTTP header propagation logic

Related to GH-422 and GH-606

Author: olegz

spring-cloud-function-samples/function-sample-cloudevent/src/main/java/io/spring/cloudevent/CloudeventDemoApplication.java

@@ -113,6 +113,15 @@ public Function<Message<SpringReleaseEvent>, SpringReleaseEvent> consumeAndProdu
 		};
 	}
 
+	@Bean
+	public Function<Map<String, Object>, Map<String, Object>> consumeAndProduceCloudEventAsPojoToPojo() {
+		return ceMessage -> {
+			ceMessage.put("version", "10.0");
+			ceMessage.put("releaseDate", "01-10-2050");
+			return ceMessage;
+		};
+	}
+
 	@Bean
 	public CloudEventAttributesProvider cloudEventAttributesProvider() {
 		return new CustomCloudEventAtttributesProvider();

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/RequestProcessor.java

@@ -191,7 +191,24 @@ private Mono<ResponseEntity<?>> response(FunctionWrapper request, Object handler
 					.map(message -> message.getPayload());
 		}
 		else {
-			builder.headers(HeaderUtils.sanitize(request.headers()));
+			if (result instanceof Mono) {
+				result = Mono.from(result)
+				.map(message -> MessageUtils.unpack(handler, message))
+				.doOnNext(value -> {
+					builder.headers(HeaderUtils.sanitize(request.headers()));
+					addHeaders(builder, value);
+				})
+				.map(message -> message.getPayload());
+			}
+			else {
+				result = Flux.from(result)
+				.map(message -> MessageUtils.unpack(handler, message))
+				.doOnNext(value -> {
+					builder.headers(HeaderUtils.sanitize(request.headers()));
+					addHeaders(builder, value);
+				})
+				.map(message -> message.getPayload());
+			}
 		}
 
 		if (isOutputSingle(handler)

spring-cloud-function-web/src/main/java/org/springframework/cloud/function/web/util/FunctionWebUtils.java

@@ -115,7 +115,7 @@ else if (result instanceof Mono) {
 		}
 		else if (result instanceof Message) {
 			if (!isMessage) {
-				result = ((Message) result).getPayload();
+//				result = ((Message) result).getPayload();
 			}
 			else if (((Message) result).getPayload() instanceof byte[]) {
 				String str = new String((byte[]) ((Message) result).getPayload());