CAMEL-6934: Xtokenize with streaming() does not close stream.

camel-core/src/main/java/org/apache/camel/support/XMLTokenExpressionIterator.java

@@ -48,7 +48,7 @@
 import org.slf4j.LoggerFactory;
 
 /**
- *
+ * An {@link org.apache.camel.language.tokenizer.XMLTokenizeLanguage} based iterator.
  */
 public class XMLTokenExpressionIterator extends ExpressionAdapter implements NamespaceAware {
     protected final String path;
@@ -89,19 +89,11 @@ public void setGroup(int group) {
     }
 
     protected Iterator<?> createIterator(InputStream in, String charset) throws XMLStreamException, UnsupportedEncodingException {
-        Reader reader;
-        if (charset == null) {
-            reader = new InputStreamReader(in);
-        } else {
-            reader = new InputStreamReader(in, charset);
-        }
-        XMLTokenIterator iterator = new XMLTokenIterator(path, nsmap, mode, group, reader);
-        return iterator;
+        return new XMLTokenIterator(path, nsmap, mode, group, in, charset);
     }
 
     protected Iterator<?> createIterator(Reader in) throws XMLStreamException {
-        XMLTokenIterator iterator = new XMLTokenIterator(path, nsmap, mode, group, in);
-        return iterator;
+        return new XMLTokenIterator(path, nsmap, mode, group, in);
     }
 
     @Override
@@ -158,6 +150,8 @@ static class XMLTokenIterator implements Iterator<Object>, Closeable {
         private static final Logger LOG = LoggerFactory.getLogger(XMLTokenIterator.class);
         private static final Pattern NAMESPACE_PATTERN = Pattern.compile("xmlns(:\\w+|)\\s*=\\s*('[^']*'|\"[^\"]*\")");
 
+        private transient InputStream originalInputStream;
+
         private AttributedQName[] splitpath;
         private int index;
         private char mode;
@@ -182,12 +176,14 @@ public XMLTokenIterator(String path, Map<String, String> nsmap, char mode, Input
             throws XMLStreamException, UnsupportedEncodingException {
             // woodstox's getLocation().etCharOffset() does not return the offset correctly for InputStream, so use Reader instead.
             this(path, nsmap, mode, 1, new InputStreamReader(in, charset));
+            this.originalInputStream = in;
         }
 
         public XMLTokenIterator(String path, Map<String, String> nsmap, char mode, int group, InputStream in, String charset) 
             throws XMLStreamException, UnsupportedEncodingException {
             // woodstox's getLocation().etCharOffset() does not return the offset correctly for InputStream, so use Reader instead.
-            this(path, nsmap, mode, new InputStreamReader(in, charset));
+            this(path, nsmap, mode, group, new InputStreamReader(in, charset));
+            this.originalInputStream = in;
         }
 
         public XMLTokenIterator(String path, Map<String, String> nsmap, char mode, Reader in) throws XMLStreamException {
@@ -278,7 +274,7 @@ private int readNext() throws XMLStreamException {
             return c;
         }
 
-        private String getCurrenText() {
+        private String getCurrentText() {
             int pos = reader.getLocation().getCharacterOffset();
             String txt = in.getText(pos - consumed);
             consumed = pos;
@@ -357,7 +353,7 @@ private String getCurrentToken() throws XMLStreamException {
             readCurrent(true);
             popName();
 
-            String token = createContextualToken(getCurrenText());
+            String token = createContextualToken(getCurrentText());
             if (mode == 'i') {
                 popNamespaces();
             }
@@ -466,7 +462,7 @@ private String getNextToken() throws XMLStreamException {
                         LOG.trace("se={}; depth={}; trackdepth={}", new Object[]{name, depth, trackdepth});
                     }
 
-                    String token = getCurrenText();
+                    String token = getCurrentText();
                     // perform the second compliance test
                     if (!compliant) {
                         if (token != null && token.startsWith("<") && !token.startsWith("<?")) {
@@ -600,8 +596,12 @@ public void remove() {
         public void close() throws IOException {
             try {
                 reader.close();
-            } catch (XMLStreamException e) {
-                throw new IOException(e);
+            } catch (Exception e) {
+                // ignore
+            }
+            // need to close the original input stream as well as the reader do not delegate close it
+            if (originalInputStream != null) {
+                IOHelper.close(originalInputStream);
             }
         }
     }

camel-core/src/test/java/org/apache/camel/language/tokenizer/XMLTokenizeLanguageStreamingFileTest.java

@@ -0,0 +1,64 @@
+/**
+ * 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.language.tokenizer;
+
+import org.apache.camel.ContextTestSupport;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.builder.xml.Namespaces;
+
+public class XMLTokenizeLanguageStreamingFileTest extends ContextTestSupport {
+
+    @Override
+    protected void setUp() throws Exception {
+        deleteDirectory("target/xmltokenize");
+        super.setUp();
+    }
+
+    public void testFromFile() throws Exception {
+        getMockEndpoint("mock:result").expectedBodiesReceived("<c:child some_attr='a' anotherAttr='a' xmlns:c=\"urn:c\"></c:child>",
+                "<c:child some_attr='b' anotherAttr='b' xmlns:c=\"urn:c\"></c:child>",
+                "<c:child some_attr='c' anotherAttr='c' xmlns:c=\"urn:c\"></c:child>",
+                "<c:child some_attr='d' anotherAttr='d' xmlns:c=\"urn:c\"></c:child>");
+
+        String body = "<?xml version='1.0' encoding='UTF-8'?>"
+                + "<c:parent xmlns:c='urn:c'>"
+                +   "<c:child some_attr='a' anotherAttr='a'></c:child>"
+                +   "<c:child some_attr='b' anotherAttr='b'></c:child>"
+                +   "<c:child some_attr='c' anotherAttr='c'></c:child>"
+                +   "<c:child some_attr='d' anotherAttr='d'></c:child>"
+                + "</c:parent>";
+
+        deleteDirectory("target/xmltokenize");
+        template.sendBodyAndHeader("file:target/xmltokenize", body, Exchange.FILE_NAME, "myxml.xml");
+
+        assertMockEndpointsSatisfied();
+    }
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            Namespaces ns = new Namespaces("C", "urn:c");
+            public void configure() {
+                from("file:target/xmltokenize")
+                    .split().xtokenize("//C:child", ns).streaming()
+                        .to("mock:result")
+                    .end();
+            }
+        };
+    }
+}

camel-core/src/test/java/org/apache/camel/language/tokenizer/XMLTokenizeLanguageStreamingTest.java

@@ -0,0 +1,37 @@
+/**
+ * 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.language.tokenizer;
+
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.apache.camel.builder.xml.Namespaces;
+
+public class XMLTokenizeLanguageStreamingTest extends XMLTokenizeLanguageTest {
+
+    @Override
+    protected RouteBuilder createRouteBuilder() {
+        return new RouteBuilder() {
+            Namespaces ns = new Namespaces("C", "urn:c");
+            public void configure() {
+                from("direct:start")
+                    .split().xtokenize("//C:child", ns).streaming()
+                        .to("mock:result")
+                    .end();
+            }
+        };
+    }
+}