CAMEL-26110: camel-as2 - do not pool a connection the partner closed

components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ClientConnection.java

@@ -16,6 +16,7 @@
  */
 package org.apache.camel.component.as2.api;
 
+import java.io.ByteArrayInputStream;
 import java.io.IOException;
 import java.net.Socket;
 import java.nio.charset.StandardCharsets;
@@ -48,10 +49,13 @@
 import org.apache.hc.client5.http.ssl.TlsSocketStrategy;
 import org.apache.hc.core5.http.*;
 import org.apache.hc.core5.http.config.Http1Config;
+import org.apache.hc.core5.http.impl.DefaultConnectionReuseStrategy;
 import org.apache.hc.core5.http.impl.io.HttpRequestExecutor;
 import org.apache.hc.core5.http.io.HttpClientConnection;
 import org.apache.hc.core5.http.io.HttpConnectionFactory;
 import org.apache.hc.core5.http.io.SocketConfig;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.entity.InputStreamEntity;
 import org.apache.hc.core5.http.message.BasicClassicHttpRequest;
 import org.apache.hc.core5.http.message.BasicHttpRequest;
 import org.apache.hc.core5.http.message.MessageSupport;
@@ -189,11 +193,6 @@ public ManagedHttpClientConnection createConnection(final Socket socket) throws
         connectionKeepAliveStrategy = (response, context) -> {
             TimeValue ttl = TimeValue.of(connectionPoolTtl);
             for (Header h : response.getHeaders()) {
-                if (AS2Header.CONNECTION.equalsIgnoreCase(h.getName())) {
-                    if (AS2Header.CLOSE.equalsIgnoreCase(h.getValue())) {
-                        ttl = TimeValue.NEG_ONE_MILLISECOND;
-                    }
-                }
                 if (AS2Header.KEEP_ALIVE.equalsIgnoreCase(h.getName())) {
                     HeaderElement headerElement = MessageSupport.parseElements(h).get(0);
                     if (headerElement.getValue() != null && "timeout".equalsIgnoreCase(headerElement.getName())) {
@@ -277,18 +276,32 @@ public ClassicHttpResponse execute(ClassicHttpRequest request, HttpClientConnect
                         "Proxy " + route.getProxyHost().toHostString() + " refused to forward the request to "
                                         + targetHost.toHostString() + ": " + new StatusLine(response));
             }
+            if (response instanceof ClassicHttpResponse classicResponse && classicResponse.getEntity() != null
+                    && classicResponse.getEntity().isStreaming()) {
+                // read the body while the connection is ours: release may close it, or hand it to another send.
+                // A parsed entity was already read by EntityParser in receiveResponseEntity and is kept as it is.
+                // The entity stays a streaming one: EntityParser rejects any other
+                HttpEntity entity = classicResponse.getEntity();
+                try {
+                    byte[] content = EntityUtils.toByteArray(entity);
+                    classicResponse.setEntity(new InputStreamEntity(
+                            new ByteArrayInputStream(content), content.length,
+                            ContentType.parseLenient(entity.getContentType()),
+                            entity.getContentEncoding()));
+                } catch (IOException | RuntimeException e) {
+                    // a connection with a half-read body must not go back to the pool
+                    endpoint.close(CloseMode.IMMEDIATE);
+                    throw e;
+                }
+            }
         } finally {
             if (endpoint != null) {
+                // a null keep-alive makes the pool close the connection: a zero or negative one keeps it with no expiry
                 TimeValue keepAlive = null;
-                if (response != null) {
-                    // Normal happy path: release with computed keep-alive.
+                if (response != null && DefaultConnectionReuseStrategy.INSTANCE.keepAlive(request, response, httpContext)) {
+                    // the partner allows the connection to be reused (no Connection: close, the response has a length)
                     keepAlive = connectionKeepAliveStrategy.getKeepAliveDuration(response, httpContext);
                 }
-                if (keepAlive == null) {
-                    // If we didn't get a response, something failed mid-flight:
-                    // do NOT keep this connection around. Close it and release with zero keep-alive.
-                    keepAlive = TimeValue.ZERO_MILLISECONDS;
-                }
                 try {
                     connectionPoolManager.release(endpoint, null, keepAlive);
                 } catch (Exception ignored) {

components/camel-as2/camel-as2-api/src/test/java/org/apache/camel/component/as2/api/AS2ClientConnectionReuseTest.java

@@ -0,0 +1,256 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.as2.api;
+
+import java.io.BufferedInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.Locale;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import org.apache.camel.util.IOHelper;
+import org.apache.hc.core5.http.ClassicHttpResponse;
+import org.apache.hc.core5.http.ContentType;
+import org.apache.hc.core5.http.HttpException;
+import org.apache.hc.core5.http.io.entity.EntityUtils;
+import org.apache.hc.core5.http.io.entity.StringEntity;
+import org.apache.hc.core5.http.message.BasicClassicHttpRequest;
+import org.apache.hc.core5.http.protocol.HttpCoreContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+/**
+ * The client keeps a connection in its pool only when the partner's response allows it: a connection the partner closes
+ * must not be leased again, or the next send writes on a closed socket (CAMEL-26110).
+ */
+class AS2ClientConnectionReuseTest {
+
+    private static final int SENDS = 3;
+
+    private ExecutorService executor;
+    private ServerSocket serverSocket;
+    private final AtomicInteger connections = new AtomicInteger();
+
+    @AfterEach
+    void stopPartner() {
+        IOHelper.close(serverSocket);
+        if (executor != null) {
+            executor.shutdownNow();
+        }
+    }
+
+    @Test
+    void connectionClosedByThePartnerIsNotReused() throws Exception {
+        startPartner("HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 2\r\n\r\nok", true);
+
+        assertEachSendSucceeds();
+        // the sends follow each other within the pool's 2s stale check: each needs a new connection
+        assertEquals(SENDS, connections.get());
+    }
+
+    @Test
+    void bodyLargerThanTheInputBufferIsFullyReadWhenThePartnerClosesTheConnection() throws Exception {
+        // the body is far beyond the 8 KB input buffer: it must be read before the connection is released and closed
+        String body = "0123456789abcdef".repeat(4096);
+        startPartner("HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: " + body.length() + "\r\n\r\n" + body, true);
+
+        assertEachSendSucceeds(body);
+        assertEquals(SENDS, connections.get());
+    }
+
+    @Test
+    void chunkedBodyLargerThanTheInputBufferIsFullyReadWhenThePartnerClosesTheConnection() throws Exception {
+        String body = "0123456789abcdef".repeat(4096);
+        startPartner("HTTP/1.1 200 OK\r\nConnection: close\r\nTransfer-Encoding: chunked\r\n\r\n"
+                     + Integer.toHexString(body.length()) + "\r\n" + body + "\r\n0\r\n\r\n",
+                true);
+
+        assertEachSendSucceeds(body);
+        assertEquals(SENDS, connections.get());
+    }
+
+    @Test
+    void largeBodyOfAKeptAliveConnectionIsReadSoTheConnectionIsReused() throws Exception {
+        String body = "0123456789abcdef".repeat(4096);
+        startPartner("HTTP/1.1 200 OK\r\nContent-Length: " + body.length() + "\r\n\r\n" + body, false);
+
+        assertEachSendSucceeds(body);
+        assertEquals(1, connections.get());
+    }
+
+    @Test
+    void truncatedBodyFailsTheSendAndTheConnectionIsNotReused() throws Exception {
+        // the partner announces 100 bytes, sends 10 and hangs up, without saying Connection: close
+        startPartner("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789", true);
+        AS2ClientConnection clientConnection = newClientConnection();
+
+        for (int i = 0; i < 2; i++) {
+            assertThrows(IOException.class, () -> clientConnection.send(newRequest(), as2Context()));
+        }
+        assertEquals(2, connections.get());
+    }
+
+    @Test
+    void bodyThatStallsFailsTheSendAndTheConnectionIsNotReused() throws Exception {
+        // the partner sends 10 of 100 bytes and keeps the connection open: the read times out with it still open
+        startPartner("HTTP/1.1 200 OK\r\nContent-Length: 100\r\n\r\n0123456789", false);
+        AS2ClientConnection clientConnection = newClientConnection(Duration.ofSeconds(1));
+
+        for (int i = 0; i < 2; i++) {
+            assertThrows(IOException.class, () -> clientConnection.send(newRequest(), as2Context()));
+        }
+        assertEquals(2, connections.get());
+    }
+
+    @Test
+    void unknownCharsetInTheResponseFailsTheSendAsAnHttpExceptionAndTheConnectionIsNotReused() throws Exception {
+        startPartner("HTTP/1.1 200 OK\r\nContent-Type: text/plain; charset=no-such-charset\r\nContent-Length: 2\r\n\r\nok",
+                false);
+        AS2ClientConnection clientConnection = newClientConnection();
+
+        for (int i = 0; i < 2; i++) {
+            assertThrows(HttpException.class, () -> clientConnection.send(newRequest(), as2Context()));
+        }
+        assertEquals(2, connections.get());
+    }
+
+    @Test
+    void http10ConnectionWithoutKeepAliveIsNotReused() throws Exception {
+        startPartner("HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nok", true);
+
+        assertEachSendSucceeds();
+        assertEquals(SENDS, connections.get());
+    }
+
+    @Test
+    void keptAliveConnectionIsReused() throws Exception {
+        startPartner("HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok", false);
+
+        assertEachSendSucceeds();
+        assertEquals(1, connections.get());
+    }
+
+    private void assertEachSendSucceeds() throws Exception {
+        assertEachSendSucceeds("ok");
+    }
+
+    private void assertEachSendSucceeds(String expectedBody) throws Exception {
+        AS2ClientConnection clientConnection = newClientConnection();
+        for (int i = 0; i < SENDS; i++) {
+            ClassicHttpResponse response = (ClassicHttpResponse) clientConnection.send(newRequest(), as2Context());
+            assertEquals(200, response.getCode(), "send " + i);
+            // read the response like AS2ClientManager does, so a reused connection starts clean
+            assertEquals(expectedBody, EntityUtils.toString(response.getEntity()), "send " + i);
+        }
+    }
+
+    private AS2ClientConnection newClientConnection() throws IOException {
+        return newClientConnection(Duration.ofSeconds(5));
+    }
+
+    private AS2ClientConnection newClientConnection(Duration socketTimeout) throws IOException {
+        return new AS2ClientConnection(
+                "1.1", "Camel AS2 Client Endpoint", "example.org", "localhost", serverSocket.getLocalPort(),
+                socketTimeout, Duration.ofSeconds(5), 5, Duration.ofMinutes(15), null, null);
+    }
+
+    private static BasicClassicHttpRequest newRequest() {
+        BasicClassicHttpRequest request = new BasicClassicHttpRequest("POST", "/");
+        request.setEntity(new StringEntity("EDI message", ContentType.TEXT_PLAIN));
+        return request;
+    }
+
+    private static HttpCoreContext as2Context() {
+        HttpCoreContext context = HttpCoreContext.create();
+        context.setAttribute(AS2ClientManager.FROM, "mrAS@example.org");
+        context.setAttribute(AS2ClientManager.AS2_FROM, "878051556");
+        context.setAttribute(AS2ClientManager.AS2_TO, "878051556");
+        context.setAttribute(AS2ClientManager.SUBJECT, "Test Case");
+        return context;
+    }
+
+    /**
+     * A partner that gives every request the same response, and closes the connection after it when told to.
+     */
+    private void startPartner(String response, boolean closeAfterResponse) throws IOException {
+        serverSocket = new ServerSocket(0, 50, InetAddress.getLoopbackAddress());
+        executor = Executors.newCachedThreadPool();
+        executor.execute(() -> {
+            while (!serverSocket.isClosed()) {
+                try {
+                    Socket socket = serverSocket.accept();
+                    executor.execute(() -> answer(socket, response, closeAfterResponse));
+                } catch (IOException e) {
+                    // closed
+                }
+            }
+        });
+    }
+
+    private void answer(Socket socket, String response, boolean closeAfterResponse) {
+        try (socket) {
+            InputStream in = new BufferedInputStream(socket.getInputStream());
+            OutputStream out = socket.getOutputStream();
+            boolean firstRequest = true;
+            do {
+                long contentLength = -1;
+                String line = readLine(in);
+                if (line == null) {
+                    // the client's startup check connects without sending a request
+                    return;
+                }
+                if (firstRequest) {
+                    connections.incrementAndGet();
+                    firstRequest = false;
+                }
+                for (line = readLine(in); line != null && !line.isEmpty(); line = readLine(in)) {
+                    if (line.toLowerCase(Locale.ROOT).startsWith("content-length:")) {
+                        contentLength = Long.parseLong(line.substring(line.indexOf(':') + 1).trim());
+                    }
+                }
+                in.skipNBytes(Math.max(contentLength, 0));
+                out.write(response.getBytes(StandardCharsets.ISO_8859_1));
+                out.flush();
+            } while (!closeAfterResponse);
+        } catch (IOException e) {
+            // connection closed
+        }
+    }
+
+    private static String readLine(InputStream in) throws IOException {
+        ByteArrayOutputStream line = new ByteArrayOutputStream();
+        int b;
+        while ((b = in.read()) != -1 && b != '\n') {
+            if (b != '\r') {
+                line.write(b);
+            }
+        }
+        return b == -1 && line.size() == 0 ? null : line.toString(StandardCharsets.ISO_8859_1);
+    }
+}