Camel c.a.存储.公共.存储输入流：流已关闭

xdnvmnnf  于 2022-11-07  发布在  Apache
关注(0)|答案(1)|浏览(101)

我正在使用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)之前使用这些头设置。

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.存储器.公共.存储器输入流：流已关闭。”

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]
apache-camel

1条答案

按热度按时间
cuxqih21

cuxqih211#

在处理器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()
赞(0）分享  回复(0）举报  2022-11-07
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com