azure - c.a.storage.common.Storage输入流: Stream is already closed

Question

我正在使用camel-azure-storage-blob在azure blob存储中移动文件。我正在尝试执行以下路由配置。

                from(azurebloburi)
                        .process(getHeaders())
                        .log(LoggingLevel.INFO, log, "After adding Headers =${in.headers})")
                        .doTry()
                        .to(myapiendpoint)
                        .process(moveTheFile(SUCCESS_CONTAINER))
                        .toD(mysuccesscontaineruri-azure)
                        .doCatch(Exception.class)
                        .log(LoggingLevel.ERROR, log,
                                "Exception - Moving into ERROR.")
                        .process(moveTheFile(ERROR_CONTAINER))
                        .convertBodyTo(byte[].class)
                        .toD(myerrorcontainer-azure)
                        .end()

我在处理文件 .toD(ERROR_CONTAINER) 之前使用这些 header 设置。

exchange.getIn().setHeader(BlobConstants.BLOB_NAME, "changingBlobName");
                exchange.getIn().setHeader(BlobConstants.BLOB_OPERATION,BlobOperationsDefinition.uploadBlockBlob);
                exchange.getIn().setHeader(BlobConstants.BLOB_CONTAINER_NAME, "myexistingsourcelistener");

我收到以下错误。任何建议都会有帮助。

“c.a.storage.common.StorageInputStream:流已关闭。”

org.apache.camel.InvalidPayloadException: No body available of type: byte[] but has value: com.azure.storage.blob.specialized.BlobInputStream@5e692e22 of type: com.azure.storage.blob.specialized.BlobInputStream on: Message. Caused by: Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.. Exchange[E8E38A893F8509F-0000000000000000]. Caused by: [org.apache.camel.TypeConversionException - Error during type conversion from type: com.azure.storage.blob.specialized.BlobInputStream to the required type: byte[] with value com.azure.storage.blob.specialized.BlobInputStream@5e692e22 due to java.lang.RuntimeException: Stream is already closed.]
    at org.apache.camel.support.MessageSupport.getMandatoryBody(MessageSupport.java:125) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:106) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.support.processor.ConvertBodyProcessor.process(ConvertBodyProcessor.java:144) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

已编辑

如果我删除 .convertBodyTo(byte[].class) 然后出现以下错误。

java.lang.RuntimeException: Stream is already closed.
    at com.azure.storage.common.StorageInputStream.checkStreamState(StorageInputStream.java:135) ~[azure-storage-common-12.11.1.jar:na]
    at com.azure.storage.common.StorageInputStream.readInternal(StorageInputStream.java:311) ~[azure-storage-common-12.11.1.jar:na]
    at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:289) ~[azure-storage-common-12.11.1.jar:na]
    at com.azure.storage.common.StorageInputStream.read(StorageInputStream.java:234) ~[azure-storage-common-12.11.1.jar:na]
    at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2314) ~[commons-io-2.6.jar:2.6]
    at org.apache.commons.io.IOUtils.copy(IOUtils.java:2270) ~[commons-io-2.6.jar:2.6]
    at org.apache.commons.io.IOUtils.copyLarge(IOUtils.java:2291) ~[commons-io-2.6.jar:2.6]
    at org.apache.commons.io.IOUtils.copy(IOUtils.java:2246) ~[commons-io-2.6.jar:2.6]
    at org.apache.commons.io.IOUtils.toByteArray(IOUtils.java:765) ~[commons-io-2.6.jar:2.6]
    at org.apache.camel.component.azure.storage.blob.BlobUtils.getInputStreamLength(BlobUtils.java:37) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobStreamAndLength.createBlobStreamAndLengthFromExchangeBody(BlobStreamAndLength.java:50) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.operations.BlobOperations.uploadBlockBlob(BlobOperations.java:181) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobProducer.process(BlobProducer.java:86) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.support.AsyncProcessorConverterHelper$ProcessorToAsyncProcessorBridge.process(AsyncProcessorConverterHelper.java:66) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.SendDynamicProcessor.lambda$process$0(SendDynamicProcessor.java:197) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.support.cache.DefaultProducerCache.doInAsyncProducer(DefaultProducerCache.java:318) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.SendDynamicProcessor.process(SendDynamicProcessor.java:182) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.Pipeline$PipelineTask.run(Pipeline.java:109) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor$Worker.schedule(DefaultReactiveExecutor.java:179) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.DefaultReactiveExecutor.scheduleMain(DefaultReactiveExecutor.java:64) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.processor.Pipeline.process(Pipeline.java:184) ~[camel-core-processor-3.11.3.jar:3.11.3]
    at org.apache.camel.impl.engine.CamelInternalProcessor.process(CamelInternalProcessor.java:399) ~[camel-base-engine-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobConsumer.processBatch(BlobConsumer.java:156) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.component.azure.storage.blob.BlobConsumer.poll(BlobConsumer.java:71) ~[camel-azure-storage-blob-3.11.3.jar:3.11.3]
    at org.apache.camel.support.ScheduledPollConsumer.doRun(ScheduledPollConsumer.java:192) ~[camel-support-3.11.3.jar:3.11.3]
    at org.apache.camel.support.ScheduledPollConsumer.run(ScheduledPollConsumer.java:107) ~[camel-support-3.11.3.jar:3.11.3]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java:305) ~[na:na]
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:834) ~[na:na]

Answer 1

在处理器之前使用流缓存 BlobInputStream 是经典 Java InputStream 实现。

.log(LoggingLevel.ERROR, log,
                            "Exception - Moving into ERROR.")
                            .streamCaching()
                            .to("direct:test")
                            from("direct:test")
                    .process(moveTheFile(ERROR_CONTAINER))
                    .convertBodyTo(byte[].class)
                    .toD(myerrorcontainer-azure)
                    .end()