我正在使用Spring Integration 4.2.5.RELEASE和Spring集成Java DSL 1.1.2.RELEASE。LambdaMessageProcessor无法识别ConversionService的有效负载类型
我无法使自定义转换工作。我已经注册自定义转换器转换成从byte[]
到my.object.MyClass
@Bean
@IntegrationConverter
public Converter bytesToMyClass() {
return new Converter<byte[], my.object.MyClass>() {
@Override
public my.object.MyClass convert(byte[] source) {
try {
return my.object.MyClass.newBuilder().mergeFrom(source).build();
} catch (InvalidProtocolBufferException e) {
throw new RuntimeException("Could not convert message.", e);
}
}
};
}
然后,我建立了我的集成流程。目标是根据名为insertBuffer
的变量将消息路由到两条路径中的一条。如果insertBuffer
> 1,则聚合消息。否则,只需将单个消息包装在一个集合中,并将其发送到服务方法。这里是我的流程:
@Bean
public IntegrationFlow routeInput(MessageChannel input, MyClassService service) {
return IntegrationFlows.from(input)
.<my.object.MyClass, Boolean>route((my.object.MyClass payload) -> insertBuffer > 1, mapping -> mapping
.subFlowMapping("true", aggregateflow -> aggregateflow
.<my.object.MyClass, Collection<my.object.MyClass>>aggregate(a -> a
.correlationStrategy(message -> 0) //all messages are part of the same group for now.
.releaseStrategy(group -> group.size() >= insertBuffer)
.sendPartialResultOnExpiry(true)
.expireGroupsUponCompletion(true)
.expireGroupsUponTimeout(true)
.groupTimeout(2000)))
.subFlowMapping("false", single -> single
.<my.object.MyClass, Collection<my.object.MyClass>>transform(Arrays::asList)
))
.handle(Collection.class, (payload, headers) ->
service.saveResult(payload))
.get();
}
然而,当我尝试运行此,我得到(以下全栈)java.lang.ClassCastException: [B cannot be cast to my.object.MyClass
以下情况例外。
一些调试后,我注意到,当org.springframework.integration.dsl.LambdaMessageProcessor#processMessage
试图处理该消息,payloadType
是java.lang.Object
时,我觉得应该是my.object.MyClass
。
看来我的所有仿制药都是正确的,我错过了什么?
完整堆栈跟踪:
ERROR o.s.i.handler.LoggingHandler - org.springframework.messaging.MessageHandlingException: nested exception is java.lang.ClassCastException: [B cannot be cast to my.object.MyClass
at org.springframework.integration.dsl.LambdaMessageProcessor.processMessage(LambdaMessageProcessor.java:130)
at org.springframework.integration.router.AbstractMessageProcessingRouter.getChannelKeys(AbstractMessageProcessingRouter.java:80)
at org.springframework.integration.router.AbstractMappingMessageRouter.determineTargetChannels(AbstractMappingMessageRouter.java:148)
at org.springframework.integration.router.AbstractMessageRouter.handleMessageInternal(AbstractMessageRouter.java:154)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.dispatcher.AbstractDispatcher.tryOptimizedDispatch(AbstractDispatcher.java:116)
at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:147)
at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:120)
at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:442)
at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:392)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:115)
at org.springframework.messaging.core.GenericMessagingTemplate.doSend(GenericMessagingTemplate.java:45)
at org.springframework.messaging.core.AbstractMessageSendingTemplate.send(AbstractMessageSendingTemplate.java:105)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutput(AbstractMessageProducingHandler.java:231)
at org.springframework.integration.handler.AbstractMessageProducingHandler.produceOutput(AbstractMessageProducingHandler.java:154)
at org.springframework.integration.handler.AbstractMessageProducingHandler.sendOutputs(AbstractMessageProducingHandler.java:102)
at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:105)
at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:127)
at org.springframework.integration.endpoint.PollingConsumer.handleMessage(PollingConsumer.java:103)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.doPoll(AbstractPollingEndpoint.java:251)
at org.springframework.integration.endpoint.AbstractPollingEndpoint.access$000(AbstractPollingEndpoint.java:57)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:176)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$1.call(AbstractPollingEndpoint.java:173)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller$1.run(AbstractPollingEndpoint.java:330)
at org.springframework.integration.util.ErrorHandlingTaskExecutor$1.run(ErrorHandlingTaskExecutor.java:55)
at org.springframework.core.task.SyncTaskExecutor.execute(SyncTaskExecutor.java:50)
at org.springframework.integration.util.ErrorHandlingTaskExecutor.execute(ErrorHandlingTaskExecutor.java:51)
at org.springframework.integration.endpoint.AbstractPollingEndpoint$Poller.run(AbstractPollingEndpoint.java:324)
at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54)
at org.springframework.scheduling.concurrent.ReschedulingRunnable.run(ReschedulingRunnable.java:81)
at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
at java.util.concurrent.FutureTask.run(FutureTask.java:266)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:180)
at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:293)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617)
at java.lang.Thread.run(Thread.java:745)
Caused by: java.lang.ClassCastException: [B cannot be cast to my.object.MyClass
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
at java.lang.reflect.Method.invoke(Method.java:497)
at org.springframework.integration.dsl.LambdaMessageProcessor.processMessage(LambdaMessageProcessor.java:127)
... 37 more
这让我更进一步。谢谢。对聚合器有何建议?它似乎只是聚合'byte []',我希望它先做转换,然后聚合。那可能吗?我需要一个定制处理器来做到这一点吗? – FGreg
M-m-m。一个新的SO问题? –
呃。我只是在聚集器前面加了一个'.transform(my.object.MyClass.class,m - > m)'。有点违反直觉,但足够好。 – FGreg