6.X: drain message body on request/response replacement (#2715)

Author: christiangoerdes

core/src/main/java/com/predic8/membrane/core/exchange/AbstractExchange.java

@@ -108,6 +108,9 @@ public Request getRequest() {
 	}
 
 	public void setRequest(Request request) {
+
+		if (this.request != null) this.request.discardBody(); // Drain the previous message body to avoid unread bytes on keep-alive connections.
+
 		this.request = request;
 		if (this.request != null) {
 			this.request.setErrorMessage(errMessage);
@@ -123,6 +126,9 @@ public Response getResponse() {
 	}
 
 	public void setResponse(Response res) {
+
+		if (response != null) response.discardBody(); // Drain the previous message body to avoid unread bytes on keep-alive connections.
+
 		response = res;
 		if (response != null) {
 			response.setErrorMessage(errMessage);

core/src/test/java/com/predic8/membrane/core/exchange/AbstractExchangeTest.java

@@ -0,0 +1,59 @@
+package com.predic8.membrane.core.exchange;
+
+import com.predic8.membrane.core.http.Body;
+import com.predic8.membrane.core.http.Message;
+import com.predic8.membrane.core.http.Request;
+import com.predic8.membrane.core.http.Response;
+import org.junit.jupiter.api.Test;
+
+import java.io.ByteArrayInputStream;
+import java.util.function.BiConsumer;
+import java.util.function.Function;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+class AbstractExchangeTest {
+
+    private static Body body(String s) {
+        return new Body(new ByteArrayInputStream(s.getBytes()), s.getBytes().length);
+    }
+
+    private static <M extends Message> void assertOldBodyDrainedOnReplace(Function<Body, M> msgFactory, BiConsumer<Exchange, M> setter) {
+        Exchange exc = new Exchange(null);
+
+        Body oldBody = body("old");
+        M m1 = msgFactory.apply(oldBody);
+        setter.accept(exc, m1);
+
+        assertFalse(oldBody.isRead());
+
+        Body newBody = body("new");
+        M m2 = msgFactory.apply(newBody);
+        setter.accept(exc, m2);
+
+        assertTrue(oldBody.isRead(), "old body must be read/drained when message is replaced");
+    }
+
+    @Test
+    void setResponse_readsOldBodyOnReplace() {
+        assertOldBodyDrainedOnReplace(
+                b -> {
+                    Response r = Response.ok().build();
+                    r.setBody(b);
+                    return r;
+                },
+                Exchange::setResponse
+        );
+    }
+
+    @Test
+    void setRequest_readsOldBodyOnReplace() {
+        assertOldBodyDrainedOnReplace(
+                b -> new Request() {{setBody(b);}},
+                Exchange::setRequest
+        );
+    }
+
+
+}