The jetty producer has fundamental flaws and is incapable of streaming

http requests to endpoints; it relies on an easily-exceeded 2MB buffer.

The http4 component supports streaming out of the box, but Exchanges
that originate from http requests automatically close streams when their
lifecycle ends. This means that certain Camel constructs that terminate
exchanges (e.g. recipientList, enrich, etc) must be avoided. This is an
undocumented gotcha in Camel.

Resolves fcrepo4-labs#76

Squashed commit of the following:

commit f2c7250bb1545be7aeb171d504340d2af7795482
Author: Aaron Birkland <[email protected]>
Date:   Sat Nov 12 23:28:40 2016 -0500

    Refactor intercept routes for streaming

    Http-based exchanges in streaming mode automatically close their
    connections after theexchange is done.  As the "loop" and
    "enhance" nodes result in the termination of exchanges, the
    ssubsequent closing of the stream results in "attempt to read
    from closed stream" exceptions

commit be18a35d271d6fe2d7640c57eb168257bb1cc53d
Author: Aaron Birkland <[email protected]>
Date:   Thu Nov 10 19:24:16 2016 -0500

    Complete conversion to http4

commit c665825763bd060681e3fd5222d22d6daab74a9b
Author: Aaron Birkland <[email protected]>
Date:   Thu Nov 10 16:34:51 2016 -0500

    checkstyle

commit 5023908
Author: Elliot Metsger <[email protected]>
Date:   Wed Nov 9 11:36:26 2016 -0500

    Work around limitations with buffering in the jetty component by using http4 instead.  This problem manifests itself when API-X proxies the retrieval of large (>2MiB) resources from Fedora. (fcrepo4-labs#77)
    - Use http4 component instead of jetty in the 'execute-intercept' route
    - Includes IT demonstrating the issue whe the jetty component is used

Author: birkland

fcrepo-api-x-integration/pom.xml

@@ -287,6 +287,13 @@
       <scope>test</scope>
     </dependency>
 
+    <dependency>
+      <groupId>commons-io</groupId>
+      <artifactId>commons-io</artifactId>
+      <version>2.5</version>
+      <scope>test</scope>
+    </dependency>
+
     <!-- Fedora -->
     <dependency>
       <groupId>org.fcrepo</groupId>

fcrepo-api-x-integration/src/test/java/org/fcrepo/apix/integration/InterceptingModalityIT.java

@@ -144,14 +144,19 @@ public void incomingInterceptHeaderResponseTest() throws Exception {
         onServiceRequest(ex -> {
             if (MODALITY_INTERCEPT_INCOMING.equals(ex.getIn().getHeader(HTTP_HEADER_MODALITY))) {
                 ex.getOut().setHeader("foo", "bar");
+                ex.getOut().setBody(ex.getIn().getBody());
+            } else {
+                ex.setOut(ex.getIn());
             }
         });
 
         final URI object = postFromTestResource("objects/object_InterceptingServiceIT.ttl",
                 objectContainer_intercept);
 
-        final FcrepoResponse response = client.get(object).accept("application/n-triples").perform();
-        assertTrue(IOUtils.toString(response.getBody(), "UTF-8").contains(TEST_OBJECT_TYPE));
+        try (final FcrepoResponse response = client.get(object).accept("application/n-triples").perform()) {
+            final String body = IOUtils.toString(response.getBody(), "UTF-8");
+            assertTrue(body.contains(TEST_OBJECT_TYPE));
+        }
     }
 
     @Test
@@ -167,6 +172,9 @@ public void incomingRequestBodyTest() throws Exception {
         onServiceRequest(ex -> {
             if (MODALITY_INTERCEPT_INCOMING.equals(ex.getIn().getHeader(HTTP_HEADER_MODALITY))) {
                 ex.getOut().setBody(String.format("<> a <%s> .", TYPE));
+                ex.getOut().setHeaders(ex.getIn().getHeaders());
+            } else {
+                ex.setOut(ex.getIn());
             }
         });
 
@@ -192,6 +200,7 @@ public void outgoingHeaderTest() throws Exception {
 
         // Give our request a specific body
         onServiceRequest(ex -> {
+            ex.setOut(ex.getIn());
             if (MODALITY_INTERCEPT_OUTGOING.equals(ex.getIn().getHeader(HTTP_HEADER_MODALITY))) {
                 ex.getOut().setHeader(TEST_HEADER, TEST_HEADER_VALUE);
             }
@@ -240,6 +249,7 @@ public void outgoingErrorTest() throws Exception {
 
         // Give our request a specific body
         onServiceRequest(ex -> {
+            ex.setOut(ex.getIn());
             if (MODALITY_INTERCEPT_OUTGOING.equals(ex.getIn().getHeader(HTTP_HEADER_MODALITY))) {
                 ex.getOut().setHeader(Exchange.HTTP_RESPONSE_CODE, 418);
             }
@@ -248,14 +258,10 @@ public void outgoingErrorTest() throws Exception {
         final URI object = postFromTestResource("objects/object_InterceptingServiceIT.ttl",
                 objectContainer_intercept);
 
-        final FcrepoResponse response = client.get(object).accept("application/n-triples").perform();
-
-        // Response code should be Fedora's 200
-        assertEquals(200, response.getStatusCode());
-
-        // Body should be returned as normal
-        assertTrue(IOUtils.toString(response.getBody(), "UTF-8").contains(TEST_OBJECT_TYPE));
-
+        try {
+            client.get(object).accept("application/n-triples").perform();
+        } catch (final FcrepoOperationFailedException e) {
+            assertEquals(418, e.getStatusCode());
+        }
     }
-
 }

fcrepo-api-x-integration/src/test/java/org/fcrepo/apix/integration/KarafIT.java

@@ -30,6 +30,7 @@
 
 import java.io.File;
 import java.io.FileInputStream;
+import java.io.InputStream;
 import java.net.URI;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -204,10 +205,19 @@ public default Option deployFile(String path) {
      * @return the resulting WebResource
      */
     public default WebResource testResource(String path) {
+        return testResource(path, "text/turtle");
+    }
 
+    /**
+     * Get a test resource from test-classes
+     *
+     * @param path the resource path relative to {@link #testResources}
+     * @return the resulting WebResource
+     */
+    public default WebResource testResource(String path, String contentType) {
         final File file = new File(testResources, path);
         try {
-            return WebResource.of(new FileInputStream(file), "text/turtle", URI.create(FilenameUtils.getBaseName(
+            return WebResource.of(new FileInputStream(file), contentType, URI.create(FilenameUtils.getBaseName(
                     path)), null);
         } catch (final Exception e) {
             throw new RuntimeException(e);
@@ -225,6 +235,34 @@ public default URI postFromTestResource(final String filePath, final URI intoCon
         }
     }
 
+    public default URI postFromTestResource(final String filePath, final URI intoContainer, final String contentType)
+            throws Exception {
+        return postFromTestResource(filePath, intoContainer, contentType,
+                String.format("%s_%s", testMethodName(), getBaseName(filePath)));
+    }
+
+    public default URI postFromTestResource(final String filePath, final URI intoContainer,
+                                            final String contentType, final String slug) throws Exception {
+        try (final WebResource object = testResource(filePath, contentType);
+             final FcrepoResponse response = client.post(intoContainer)
+                     .body(object.representation(), object.contentType())
+                     .slug(slug)
+                     .perform()) {
+            return response.getLocation();
+        }
+    }
+
+    public default URI postFromStream(final InputStream in, final URI intoContainer, final String contentType,
+                                      final String slug) throws Exception {
+        try (final WebResource object = WebResource.of(in, contentType);
+             final FcrepoResponse response = client.post(intoContainer)
+                     .body(object.representation(), object.contentType())
+                     .slug(slug)
+                     .perform()) {
+            return response.getLocation();
+        }
+    }
+
     public static <T> T attempt(final int times, final Callable<T> it) {
 
         Exception caught = null;

fcrepo-api-x-integration/src/test/java/org/fcrepo/apix/integration/LoaderIT.java

@@ -144,6 +144,7 @@ public List<Option> additionalKarafConfig() {
     public void setUp() {
 
         onServiceRequest(ex -> {
+            ex.setOut(ex.getIn());
             if ("OPTIONS".equals(ex.getIn().getHeader(Exchange.HTTP_METHOD))) {
                 ex.getOut().setBody(optionsResponse.get());
             } else if ("GET".equals(ex.getIn().getHeader(Exchange.HTTP_METHOD))) {

fcrepo-api-x-integration/src/test/java/org/fcrepo/apix/integration/ServiceBasedTest.java

@@ -135,11 +135,12 @@ public void configure() throws Exception {
                 from("jetty:" + serviceEndpoint +
                         "?matchOnUriPrefix=true")
                                 .process(ex -> {
-                                    ex.getOut().copyFrom(responseFromService);
                                     requestToService.copyFrom(ex.getIn());
 
                                     if (processFromTest != null) {
                                         processFromTest.process(ex);
+                                    } else {
+                                        ex.getOut().copyFrom(responseFromService);
                                     }
 
                                 });