Mehrere Datei-E/A mit Apache Kamel

Question

Meine Anforderung lautet wie folgt: Ich muss eine Datei entpacken und Metadaten-Informationen aus einer entpackten XML-Datei lesen. Dies würde in der xmlToObjectMapper getan werden. Jetzt muss ich die Metadaten mit Daten aus mehreren anderen xmls anreichern, die sich in Zip-Dateien befinden. Das heißt, ich muss sie auspacken, sie lesen und die Informationen an mein MetadataHolder Objekt anhängen. Ich wollte dies in der kommentierten Zeile pollEnrich implementieren.

from("file://{{first.directory}}?noop=true&idempotent=true") 
     .split(new ZipSplitter()) 
      .streaming() 
       .choice() 
        .when(header("zipFileName").isEqualTo("metadata.xml")) 
         .multicast() 
          .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
          .pipeline() 
           .convertBodyTo(String.class, StandardCharsets.ISO_8859_1.name()) 
           .bean("xmlToObjectMapper", "readAllRequiredMetadata") 
           //.pollEnrich("", 0, new MetadataHolderAggregationStrategy()) 
          .end() 
         .end() 
       .end() 
      .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
     .end() 
    .end() 
    ; 

Aber jetzt steckte ich bin, weil ich die folgende stacktrace erhalten:

org.apache.camel.InvalidPayloadException: No body available of type: java.lang.String but has value: org.apache.camel.dataformat.zipfile.ZipInputStreamWrapper@2c52254b of type: org.apache.camel.dataformat.zipfile.ZipInpu 
tStreamWrapper on: Message[id]. Caused by: Error during type conversion from type: java.lang.String to the required type: java.lang.String with value [Body is instance of java.io.I 
nputStream] due java.io.IOException: Stream closed. Exchange[id]. Caused by: [org.apache.camel.TypeConversionException - Error during type conversion from type: java.lang.String to 
the required type: java.lang.String with value [Body is instance of java.io.InputStream] due java.io.IOException: Stream closed] 

Ich dachte, dass Multicast implizit den Strom zwischengespeichert werden und es an alle Endpunkte senden ...

So können jemand sagt mir, warum das passiert? Und ist meine Lösung der Anforderung ein gültiger Weg oder muss ich die Anforderung in mehrere Routen aufteilen?

EDIT1:

löste ich das Problem mit der Ausnahme von .streamCaching() auf die Route hinzufügen. Wegen anderer Probleme musste ich die ursprüngliche Route ein wenig ändern.

from("file://{{first.directory}}?noop=true&idempotent=true") 
     .streamCaching() 
     .split(new ZipSplitter()) 
      .streaming() 
       .choice() 
        .when(header("zipFileName").isEqualTo("metadata.xml")) 
         .multicast() 
          .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
          .pipeline() 
           .convertBodyTo(String.class, StandardCharsets.ISO_8859_1.name()) 
           .bean("xmlToObjectMapper", "readAllRequiredMetadata") 
           .pollEnrich("file://{{second.directory}}?&noop=true&idempotent=true", 0, new MetadataHolderAggregationStrategy()) 
          .end() 
         .end() 
        .endChoice() 
        .otherwise() 
         .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
       .end() 
     .end() 
    .end() 
    ; 

Aber jetzt habe ich Probleme mit pollEnrich. Das neue Exchange in meinem MetadataHolderAggregationStrategy ist immer null. Wie kann ich das lösen?

EDIT2:

Schließlich fand ich eine Lösung. Ich musste enrich anstelle von pollEnrich verwenden und die gesamte I/O in Java implementieren (readAllRequiredMetadata2). Meine endgültige Lösung sieht so aus:

from("file://{{first.directory}}?noop=true&idempotent=true") 
     .streamCaching() 
     .split(new ZipSplitter()) 
      .streaming() 
       .choice() 
        .when(header("zipFileName").isEqualTo("metadata.xml")) 
         .multicast() 
          .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
          .pipeline() 
           .convertBodyTo(String.class, StandardCharsets.ISO_8859_1.name()) 
           .bean("xmlToObjectMapper", "readAllRequiredMetadata") 
           .enrich("direct:readResources", 0, new MetadataHolderAggregationStrategy()) 
           //Go on processing the enriched metadata 
          .end() 
         .end() 
        .endChoice() 
        .otherwise() 
         .to("file://{{first.directory}}/Working?fileName=$simple{file:onlyname.noext}/${header.zipFileName}") 
       .end() 
     .end() 
    .end() 
    ; 

from("direct:readResources") 
    .bean("xmlToObjectMapper", "readAllRequiredMetadata2") 
.end() 
; 

Answer 1

Für Stream-ähnliche Daten müssen Sie wahrscheinlich Stream-Caching verwenden. Es ist standardmäßig deaktiviert, so dass der zweite Endpunkt es nicht hat, wie es bereits vom ersten gelesen wurde. Sie können weitere Details dort überprüfen: http://camel.apache.org/stream-caching.html