Использование SpringXD почты Источник с именованными каналами

Question

, когда я создаю поток и использовать HelloSpringXDTasklet example Job так: stream create email-stream --definition "mail ... > queue:job:helloSpringXD"

Я всегда получаю исключение:

2016-05-24T10:42:29+0200 1.3.1.RELEASE ERROR task-scheduler-5 handler.LoggingHandler - org.springframework.integration.transformer.MessageTransformationExceptio 
n: Failed to transform Message; nested exception is org.springframework.messaging.MessageHandlingException: nested exception is java.lang.IllegalArgumentExcepti 
on: Unable to convert provided JSON to Map<String, Object> 
     at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:95) 
     at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:99) 
     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:471) 
     at java.util.concurrent.FutureTask.run(FutureTask.java:262) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:178) 
     at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:292) 
     at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145) 
     at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615) 
     at java.lang.Thread.run(Thread.java:745) 
Caused by: org.springframework.messaging.MessageHandlingException: nested exception is java.lang.IllegalArgumentException: Unable to convert provided JSON to Ma 
p<String, Object> 
     at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:96) 
     at org.springframework.integration.transformer.AbstractMessageProcessingTransformer.transform(AbstractMessageProcessingTransformer.java:90) 
     at org.springframework.integration.transformer.MessageTransformingHandler.handleRequestMessage(MessageTransformingHandler.java:89) 
     ... 35 more 
Caused by: java.lang.IllegalArgumentException: Unable to convert provided JSON to Map<String, Object> 
     at org.springframework.xd.dirt.plugins.job.ExpandedJobParametersConverter.getJobParametersForJsonString(ExpandedJobParametersConverter.java:169) 
     at org.springframework.xd.dirt.plugins.job.JobLaunchRequestTransformer.toJobLaunchRequest(JobLaunchRequestTransformer.java:128) 
     at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) 
     at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57) 
     at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) 
     at java.lang.reflect.Method.invoke(Method.java:606) 
     at org.springframework.expression.spel.support.ReflectiveMethodExecutor.execute(ReflectiveMethodExecutor.java:113) 
     at org.springframework.expression.spel.ast.MethodReference.getValueInternal(MethodReference.java:102) 
     at org.springframework.expression.spel.ast.MethodReference.access$000(MethodReference.java:49) 
     at org.springframework.expression.spel.ast.MethodReference$MethodValueRef.getValue(MethodReference.java:347) 
     at org.springframework.expression.spel.ast.CompoundExpression.getValueInternal(CompoundExpression.java:88) 
     at org.springframework.expression.spel.ast.SpelNodeImpl.getTypedValue(SpelNodeImpl.java:131) 
     at org.springframework.expression.spel.standard.SpelExpression.getValue(SpelExpression.java:330) 
     at org.springframework.integration.util.AbstractExpressionEvaluator.evaluateExpression(AbstractExpressionEvaluator.java:166) 
     at org.springframework.integration.util.MessagingMethodInvokerHelper.processInternal(MessagingMethodInvokerHelper.java:317) 
     at org.springframework.integration.util.MessagingMethodInvokerHelper.process(MessagingMethodInvokerHelper.java:155) 
     at org.springframework.integration.handler.MethodInvokingMessageProcessor.processMessage(MethodInvokingMessageProcessor.java:93) 
     ... 37 more 
Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'This': was expecting ('true', 'false' or 'null') 
at [Source: java.io.StringReader@4135c5a7; line: 1, column: 5] 
     at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:1419) 
     at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:508) 
     at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._reportInvalidToken(ReaderBasedJsonParser.java:2300) 
     at com.fasterxml.jackson.core.json.ReaderBasedJsonParser._handleOddValue(ReaderBasedJsonParser.java:1459) 
     at com.fasterxml.jackson.core.json.ReaderBasedJsonParser.nextToken(ReaderBasedJsonParser.java:683) 
     at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:3105) 
     at com.fasterxml.jackson.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3051) 
     at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2175) 
     at org.springframework.xd.dirt.plugins.job.ExpandedJobParametersConverter.getJobParametersForJsonString(ExpandedJobParametersConverter.java:166) 
     ... 53 more 

Мои job-definition.xml:

<bean id="helloSpringXDTasklet" 
     class="org.springframework.springxd.samples.batch.HelloSpringXDTasklet" /> 

<batch:job id="EmailImporterJob"> 
    <batch:step id="helloSpringXDStep" next="process-step"> 
     <batch:tasklet ref="helloSpringXDTasklet" /> 
    </batch:step> 
    <batch:step id="process-step"> 
     <batch:tasklet> 
      <batch:chunk reader="itemReader" processor="itemProcessor" writer="itemWriter" commit-interval="10" /> 
     </batch:tasklet> 
    </batch:step> 
</batch:job> 

Однако, поскольку задание должно печатать Привет, Весна XD! в журнал, но перед этим возникает исключение, мне интересно, какой JSON и который Map<String, Object> означает в сообщении об ошибке.

Есть ли способ использовать источник почты с именованными каналами?

Answer 1

Именованный канал здесь в этом пакетном задании является входящим каналом, который ожидает запроса запуска задания из исходного модуля. Таким образом, в этом случае вывод источника Mail считается параметром запуска задания для вашего пакетного задания.

Поскольку полезная нагрузка от Mail источника типа String преобразователя параметров работы предполагает как JSONString и пытается преобразовать в Map использовать его в качестве параметра работы.