Junit test for Spring Integration - Payload related issues

[clouddev]

Folks,
I have a simple Junit test that creates a spring integration message and uses a AMQP outbound channel adapter.
The integration context looks like

toRabbitMessageChannel -> OutBoundAdapter


The payload takes an object of "MyEntity.class"

MyEntity entityRequest = new MyEntity();
entityRequest.setId(1L);
entityRequest.setUri("my URI");

The Message is being sent using the following code

toRabbit.send(new GenericMessage<MyEntity>(entityRequest));

I have autowired the message channel through the follow annotation
@Autowired
@Qualifier("toRabbit")
MessageChannel toRabbit;


So the problem that I am facing is that the message goes out into my Rabbit queue but the payload is zero.

Properties
priority: 0
delivery_mode: 2
headers:
content_type: application/octet-stream
Payload 0 bytes Encoding: string

I single stepped the Junit test in the debugger and I see that the SimpleMessageConverter does not convert the object into a byte array.

In the function convertAndSend after the conversion is done
The "body" field of the "messagetoSend" object is NULL whereas the "message" field has the MyEntity object as set by the Junit test. My understanding is that the SimpleMessageConverter can take in any object and serialize it into a byte array. What am I doing wrong? Also I would like to understand on any transformation I would need to apply to get the object back on the receiving side.

[Gary Russell]

Your object must implement Serializable (and all its fields must be Serializable or transient).

[clouddev]

Yes, I did not have the object declared.
The content_type however is "application/x-java-serialized-object" which is expected. If I would like to send the object in Json? Can I attach a message converter to the outbound channel adapter? Is there an inbuilt converter? I do see that Spring AMQP provides this facility through a JsonMessageConvertor, can I do the same through SI?

[Gary Russell]

Yes; the default MessageConverter (SimpleMessageConverter) only handles byte[], String and Serializable.

To use the JsonMessageConverter, declare one as a <bean/> and provide it to the outbound adapter's <rabbit:template/> via the message-converter attribute.

[Gary Russell]

You can also use an <int:object-to-json--transformer/> upstream of the adapter if you prefer.

In both cases, you need Jackson on the classpath.

[clouddev]

I will try using the json transformer next.

Currently I am stuck trying to complete an echo service for MyEntity.class.

The Junit spring integration context now has an outbound gateway since it expects a response back from the echo service. I am using the MessageTemplate to send a SI message using the sendAndReceive API. Here's the code

MyEntity entityRequest = new MyEntity(1,"uri");
Message<?> reply = template.sendAndReceive(toRabbit, new GenericMessage<MyEntity>(entityRequest));

MyEntity myent = (MyEntity)reply.getPayload();
logger.info("MyEntity + " + myent.toString());

When I run the test, the Junit test fails with the following exception. I am not sure as to why the code is trying to perform a send on the "fromRabbit" channel when it should be doing a receive. The sending part was already completed. Am I using the wrong API to do a synchronous send and receive? I will appreciate any assistance from the experts on this topic.

org.springframework.integration.MessageDeliveryExc eption: Dispatcher has no subscribers for channel fromRabbit.
at org.springframework.integration.channel.AbstractSu bscribableChannel.doSend(AbstractSubscribableChann el.java:82)
at org.springframework.integration.channel.AbstractMe ssageChannel.send(AbstractMessageChannel.java:157)
at org.springframework.integration.channel.AbstractMe ssageChannel.send(AbstractMessageChannel.java:128)
at org.springframework.integration.core.MessagingTemp late.doSend(MessagingTemplate.java:288)
at org.springframework.integration.core.MessagingTemp late.send(MessagingTemplate.java:149)
at org.springframework.integration.handler.AbstractRe plyProducingMessageHandler.sendMessage(AbstractRep lyProducingMessageHandler.java:175)
at org.springframework.integration.handler.AbstractRe plyProducingMessageHandler.sendReplyMessage(Abstra ctReplyProducingMessageHandler.java:159)
at org.springframework.integration.handler.AbstractRe plyProducingMessageHandler.produceReply(AbstractRe plyProducingMessageHandler.java:124)
at org.springframework.integration.handler.AbstractRe plyProducingMessageHandler.handleResult(AbstractRe plyProducingMessageHandler.java:118)
at org.springframework.integration.handler.AbstractRe plyProducingMessageHandler.handleMessageInternal(A bstractReplyProducingMessageHandler.java:100)
at org.springframework.integration.handler.AbstractMe ssageHandler.handleMessage(AbstractMessageHandler. java:73)
at org.springframework.integration.dispatcher.Unicast ingDispatcher.doDispatch(UnicastingDispatcher.java :115)
at org.springframework.integration.dispatcher.Unicast ingDispatcher.dispatch(UnicastingDispatcher.java:1 02)
at org.springframework.integration.channel.AbstractSu bscribableChannel.doSend(AbstractSubscribableChann el.java:77)
at org.springframework.integration.channel.AbstractMe ssageChannel.send(AbstractMessageChannel.java:157)
at org.springframework.integration.channel.AbstractMe ssageChannel.send(AbstractMessageChannel.java:128)
at org.springframework.integration.core.MessagingTemp late.doSend(MessagingTemplate.java:288)
at org.springframework.integration.core.MessagingTemp late.doSendAndReceive(MessagingTemplate.java:318)
at org.springframework.integration.core.MessagingTemp late.sendAndReceive(MessagingTemplate.java:239)
at com.myentity.service.MyEntityMessageTest.test(MyEn tityMessageTest.java:98)



Based on the logging and some debugging, I see that the webapp returns MyEntity on the Rabbit MQ. The webapp uses an inbound gateway to receive requests and return responses on the same queue.

[Gary Russell]

You need to show us your S.I. configuration for the test. It looks like the service connected to toRabbit is trying to send to fromRabbit, which has no subscribers.

[clouddev]

here's the S.I configuration for the test.

<int:logging-channel-adapter id="loggingChannel" log-full-message="true" level="INFO" />

<int:channel id="toRabbit">
<int:interceptors>
<int:wire-tap channel="loggingChannel" />
</int:interceptors>
</int:channel>

<int:channel id="fromRabbit">
<int:interceptors>
<int:wire-tap channel="loggingChannel" />
</int:interceptors>
</int:channel>

<!-- <int-amqp:outbound-channel-adapter channel="toRabbit" amqp-template="amqpTemplate" routing-key="${rabbitmq.MyQueue.name}" /> -->

<int-amqp:outbound-gateway
request-channel="toRabbit"
amqp-template="amqpTemplate"
exchange-name="${rabbitmq.MyExchange.name}"
order="1"
reply-channel="fromRabbit"
routing-key="${rabbitmq.MyQueue.name}"
/>

[Amol Nayak]

Yes, as Gary mentioned, you have defined a reply-channel attribute. So whenever a reply comes back, it it sent to the channel specified in the reply-channel attribute. If one is not found, the reply channel in the message header is used.
Now since you have used MesagingTemplate to send and receive the message, a temporary channel is set in the reply channel header of the message sent. It,however, never receives a message as you already have the reply-channel attribute set.

To get this working, remove the reply-channel attribute and your sendAndReceive method should return the appropriate result

[clouddev]

Thanks a lot Amol, I tried your suggestion and it worked. I was actually trying to figure out a way to bind a specific reply channel through the MessageTemplate interface but I couldn't. Is there any way that I can keep the S.I context with the reply-channel and provide the reply-to queue to MessageTemplate?