As part of an adapter service we are building for a client, we need to
process XMLs received in a zip file from a third party service and post it
in JSON format to another API. Everything works fine unless we introduce
parallel processing for the XMLs. This is the code we are using to read the
XML files in the zip.
this.basicRoute
.log("picked file fileName=${file:absolute.path}")
.choice(when(ex =>
ex.getIn.getHeader(Exchange.FILE_NAME_ONLY).toString.matches(".*\\.zip"))
.log("inside zip file processing zipFileName=${file:absolute.path}")
.unmarshal(zipFile)
.split(_.getIn.getBody)
.streaming()
.when(_.getIn.getHeader(Exchange.FILE_NAME).toString.matches(".*\\.xml"))
.setBody(ex =>
InputStreamToXmlConverter.toXml(ex.getIn.getBody.asInstanceOf[InputStream]))
.choice(when(exchange =>
XmlValidator.isProperLocale(exchange.getIn.getBody.asInstanceOf[Node]))
.to("direct:toPath"))
.otherwise()
.to("file:" + Settings("some.ignoredPath")))
We tried a couple of approaches for parallel processing the files but get a
"Stream Closed" exception.
java.io.IOException: Stream closed
at java.util.zip.ZipInputStream.ensureOpen(ZipInputStream.java:66)
~[na:1.7.0_06]
at java.util.zip.ZipInputStream.read(ZipInputStream.java:181)
~[na:1.7.0_06]
at java.io.BufferedInputStream.read1(BufferedInputStream.java:273)
~[na:1.7.0_06]
at java.io.BufferedInputStream.read(BufferedInputStream.java:334)
~[na:1.7.0_06]
at sun.nio.cs.StreamDecoder.readBytes(StreamDecoder.java:283)
~[na:1.7.0_06]
at sun.nio.cs.StreamDecoder.implRead(StreamDecoder.java:325)
~[na:1.7.0_06]
at sun.nio.cs.StreamDecoder.read(StreamDecoder.java:177) ~[na:1.7.0_06]
at java.io.InputStreamReader.read(InputStreamReader.java:184)
~[na:1.7.0_06]
at java.io.BufferedReader.fill(BufferedReader.java:154) ~[na:1.7.0_06]
at java.io.BufferedReader.read(BufferedReader.java:175) ~[na:1.7.0_06]
at
scala.io.BufferedSource$$anonfun$iter$1$$anonfun$apply$mcI$sp$1.apply$mcI$sp(BufferedSource.scala:38)
~[scala-library.jar:na]
at scala.io.Codec.wrap(Codec.scala:68) ~[scala-library.jar:na]
at
scala.io.BufferedSource$$anonfun$iter$1.apply(BufferedSource.scala:38)
~[scala-library.jar:na]
at
scala.io.BufferedSource$$anonfun$iter$1.apply(BufferedSource.scala:38)
~[scala-library.jar:na]
at scala.collection.Iterator$$anon$9.next(Iterator.scala:162)
~[scala-library.jar:0.12.1]
at scala.collection.Iterator$$anon$17.hasNext(Iterator.scala:511)
~[scala-library.jar:0.12.1]
at scala.collection.Iterator$$anon$11.hasNext(Iterator.scala:327)
~[scala-library.jar:0.12.1]
at scala.io.Source.hasNext(Source.scala:226) ~[scala-library.jar:na]
at scala.collection.Iterator$class.foreach(Iterator.scala:727)
~[scala-library.jar:0.12.1]
at scala.io.Source.foreach(Source.scala:178) ~[scala-library.jar:na]
at
scala.collection.TraversableOnce$class.addString(TraversableOnce.scala:320)
~[scala-library.jar:0.12.1]
at scala.io.Source.addString(Source.scala:178) ~[scala-library.jar:na]
at
scala.collection.TraversableOnce$class.mkString(TraversableOnce.scala:286)
~[scala-library.jar:0.12.1]
at scala.io.Source.mkString(Source.scala:178) ~[scala-library.jar:na]
at
scala.collection.TraversableOnce$class.mkString(TraversableOnce.scala:288)
~[scala-library.jar:0.12.1]
at scala.io.Source.mkString(Source.scala:178) ~[scala-library.jar:na]
at
scala.collection.TraversableOnce$class.mkString(TraversableOnce.scala:290)
~[scala-library.jar:0.12.1]
at scala.io.Source.mkString(Source.scala:178) ~[scala-library.jar:na]
Any help on how to deal with this problem will be appreciated, thanks!
--
View this message in context:
http://camel.465427.n5.nabble.com/Reading-processing-xml-files-in-parallel-from-a-zip-container-tp5744555.html
Sent from the Camel - Users mailing list archive at Nabble.com.
