You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by da...@apache.org on 2024/07/03 07:28:43 UTC

(camel) branch main updated: CAMEL-20938: camel-http: Fix disableStreamCache cannot read input stream later sa http-client closes the stream. (#14711)

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 335495f88de CAMEL-20938: camel-http: Fix disableStreamCache cannot read input stream later sa http-client closes the stream. (#14711)
335495f88de is described below

commit 335495f88de0d8511db8fd86f307b5ed8e15cbb5
Author: Claus Ibsen <cl...@gmail.com>
AuthorDate: Wed Jul 3 09:28:37 2024 +0200

    CAMEL-20938: camel-http: Fix disableStreamCache cannot read input stream later sa http-client closes the stream. (#14711)
---
 .../apache/camel/component/http/HttpProducer.java  | 56 +++++++++++++---------
 .../component/http/HttpDisableStreamCacheTest.java | 19 ++------
 2 files changed, 37 insertions(+), 38 deletions(-)

diff --git a/components/camel-http/src/main/java/org/apache/camel/component/http/HttpProducer.java b/components/camel-http/src/main/java/org/apache/camel/component/http/HttpProducer.java
index 78e6b4aedd0..b8386a1b277 100644
--- a/components/camel-http/src/main/java/org/apache/camel/component/http/HttpProducer.java
+++ b/components/camel-http/src/main/java/org/apache/camel/component/http/HttpProducer.java
@@ -87,6 +87,7 @@ public class HttpProducer extends DefaultProducer {
     private static final Logger LOG = LoggerFactory.getLogger(HttpProducer.class);
 
     private static final Integer OK_RESPONSE_CODE = 200;
+    private static final int BUFFER_SIZE = 1024 * 2;
 
     private HttpClient httpClient;
     private final HttpContext httpContext;
@@ -531,6 +532,12 @@ public class HttpProducer extends DefaultProducer {
             // find the charset and set it to the Exchange
             HttpHelper.setCharsetFromContentType(contentType, exchange);
         }
+
+        if (ignoreResponseBody) {
+            // ignore response
+            return null;
+        }
+
         // if content type is a serialized java object then de-serialize it back to a Java object
         if (contentType != null && contentType.equals(HttpConstants.CONTENT_TYPE_JAVA_SERIALIZED_OBJECT)) {
             // only deserialize java if allowed
@@ -541,33 +548,36 @@ public class HttpProducer extends DefaultProducer {
                 return null;
             }
         } else {
-            if (!getEndpoint().isDisableStreamCache()) {
-                if (ignoreResponseBody) {
-                    // ignore response
-                    return null;
-                }
-                int max = getEndpoint().getComponent().getResponsePayloadStreamingThreshold();
-                if (max > 0) {
-                    // optimize when we have content-length for small sizes to avoid creating streaming objects
-                    long len = entity.getContentLength();
-                    if (len > 0 && len <= max) {
-                        int i = (int) len;
-                        byte[] arr = new byte[i];
-                        int read = 0;
-                        int offset = 0;
-                        int remain = i;
-                        while ((read = is.read(arr, offset, remain)) > 0 && remain > 0) {
-                            offset += read;
-                            remain -= read;
+            if (entity.isStreaming()) {
+                if (getEndpoint().isDisableStreamCache()) {
+                    // write to in-memory buffer
+                    ByteArrayOutputStream bos = new ByteArrayOutputStream(BUFFER_SIZE);
+                    entity.writeTo(bos);
+                    return bos.toByteArray();
+                } else {
+                    int max = getEndpoint().getComponent().getResponsePayloadStreamingThreshold();
+                    if (max > 0) {
+                        // optimize when we have content-length for small sizes to avoid creating streaming objects
+                        long len = entity.getContentLength();
+                        if (len > 0 && len <= max) {
+                            int i = (int) len;
+                            byte[] arr = new byte[i];
+                            int read;
+                            int offset = 0;
+                            int remain = i;
+                            while ((read = is.read(arr, offset, remain)) > 0 && remain > 0) {
+                                offset += read;
+                                remain -= read;
+                            }
+                            IOHelper.close(is);
+                            return arr;
                         }
-                        IOHelper.close(is);
-                        return arr;
                     }
+                    // else for bigger payloads then wrap the response in a stream cache so its re-readable
+                    return doExtractResponseBodyAsStream(is, exchange);
                 }
-                // else for bigger payloads then wrap the response in a stream cache so its re-readable
-                return doExtractResponseBodyAsStream(is, exchange);
             } else {
-                // use the response stream as-is
+                // use the response as-is
                 return is;
             }
         }
diff --git a/components/camel-http/src/test/java/org/apache/camel/component/http/HttpDisableStreamCacheTest.java b/components/camel-http/src/test/java/org/apache/camel/component/http/HttpDisableStreamCacheTest.java
index f0722c0bb9b..19cb4595405 100644
--- a/components/camel-http/src/test/java/org/apache/camel/component/http/HttpDisableStreamCacheTest.java
+++ b/components/camel-http/src/test/java/org/apache/camel/component/http/HttpDisableStreamCacheTest.java
@@ -16,11 +16,7 @@
  */
 package org.apache.camel.component.http;
 
-import java.io.InputStream;
-
 import org.apache.camel.Exchange;
-import org.apache.camel.TypeConversionException;
-import org.apache.camel.TypeConverter;
 import org.apache.camel.component.http.handler.BasicValidationHandler;
 import org.apache.hc.core5.http.impl.bootstrap.HttpServer;
 import org.apache.hc.core5.http.impl.bootstrap.ServerBootstrap;
@@ -28,9 +24,8 @@ import org.junit.jupiter.api.Test;
 
 import static org.apache.camel.http.common.HttpMethods.GET;
 import static org.apache.camel.test.junit5.TestSupport.assertIsInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
-import static org.junit.jupiter.api.Assertions.assertThrows;
 
 public class HttpDisableStreamCacheTest extends BaseHttpTest {
 
@@ -60,16 +55,10 @@ public class HttpDisableStreamCacheTest extends BaseHttpTest {
                 exchange1 -> {
                 });
 
-        InputStream is = assertIsInstanceOf(InputStream.class, exchange.getMessage().getBody());
-        assertNotNull(is);
-
-        String name = is.getClass().getName();
-        // should not be stream cache
-        assertFalse(name.contains("CachedOutputStream"));
+        byte[] arr = assertIsInstanceOf(byte[].class, exchange.getMessage().getBody());
+        assertNotNull(arr);
 
-        TypeConverter converter = context.getTypeConverter();
-        assertThrows(TypeConversionException.class, () -> converter.convertTo(String.class, exchange, is),
-                "Should have thrown an exception");
+        assertEquals("camel rocks!", context.getTypeConverter().convertTo(String.class, arr));
     }
 
 }