That's the funny thing about splitting and parallel: The last message with the split end marker is usually never the last message processed. We tried resequencing (via split index) without success, because you have to keep all messages in memory to guarantee sequencing works properly, otherwise only the elements of a certain chunk are in sequence. Thus we perform all cleanup/end task after procesing of the last messages (all repositories are empty) and this works without issues.
Jens Von meinem iPhone gesendet > Am 16.07.2016 um 04:50 schrieb sonu123 <[email protected]>: > > I have to do xml –xml transformation. Since source xml is expected to be very > big around 300mb, so I am splitting the sub-elements and processing > concurrently. As I am doing partial marshal and unmarshal, I am writing > root’s starting tag and ending tag separately. During addition of the > closing tag, I am facing an issue. Root end tag is not getting added at the > bottom but somewhere before end. Hence making the xml invalid. When I am > writing the end element some sub-elements were still being processed and > they appear after end element is already written. Is there any way, I can > wait until all threads in threadPool has finished with their work? > My route looks like this. > JAXBContext jaxbContextH = > JAXBContext.newInstance(HierarchicalLineType.class); > JaxbDataFormat jaxb = new JaxbDataFormat(jaxbContextH); > HierarchicalLineType c = new HierarchicalLineType(); > jaxb.setPartClass(c.getClass().getCanonicalName()); > jaxb.setFragment(true); > > JAXBContext jaxbContextFlat = > JAXBContext.newInstance(FlatLineType.class); > JaxbDataFormat jaxbFlat = new JaxbDataFormat(jaxbContextFlat); > FlatLineType flatLine = new FlatLineType(); > jaxbFlat.setPartClass(flatLine.getClass().getCanonicalName()); > jaxbFlat.setFragment(true); > > > > from("file:target/outbox?noop=true&fileName=source.xml") > .log("Starting to process Xml file: ${header.CamelFileName}") > .multicast().to( > "direct:fileInitiator","direct:xmlTransformation") > .end() > .log("Done processing big file: ${header.CamelFileName}"); > > > from("direct:fileInitiator").routeId("fileInitiator").bean(MyService, > "writeBeginningElement") > .to("file:target/outbox?fileName=target.xml&fileExist=Override"); > > from("direct:xmlTransformation") > .routeId("businessConversion") > .split(stax(HierarchicalLineType.class)) > .streaming().executorService(threadPool) > .bean(MyService.class, "transformFromHierarchyXMLToFlatXML") > .marshal(jaxbFlat) > .bean(MyService.class, "writeClosingTags") > .aggregate(header(Exchange.FILE_NAME_ONLY), > new StringBodyAggregator()).completionSize(750) > > .completionPredicate(header("CamelSplitComplete").isEqualTo(Boolean.TRUE)).completionTimeout(1500) > .to("file:target/outbox?fileName=target.xml&fileExist=Append"); > > > > public void fileFlatClosingTags(String body, Exchange exchange) { > boolean isComplete= (Boolean) > exchange.getProperty("CamelSplitComplete"); > String modBody=body; > if(isComplete){ > modBody=body+"</FlatLines>"; > > } > exchange.getIn().setBody(modBody); > } > > public String writeFlatBeginningElement(Exchange exchange){ > return "<FlatLines>"; > } > > Thank in advance > > > > > -- > View this message in context: > http://camel.465427.n5.nabble.com/How-to-make-aggregator-wait-until-all-threads-are-completed-tp5785167.html > Sent from the Camel - Users mailing list archive at Nabble.com.
