diff --git a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ClientConnection.java b/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ClientConnection.java index 570ffc7561e1a..daa82c30e5948 100644 --- a/components/camel-as2/camel-as2-api/src/main/java/org/apache/camel/component/as2/api/AS2ClientConnection.java +++ b/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) { diff --git a/components/camel-as2/camel-as2-api/src/test/java/org/apache/camel/component/as2/api/AS2ClientConnectionReuseTest.java b/components/camel-as2/camel-as2-api/src/test/java/org/apache/camel/component/as2/api/AS2ClientConnectionReuseTest.java new file mode 100644 index 0000000000000..08430ee43998d --- /dev/null +++ b/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); + } +}