Passing Object containing OutputStream to message channel

[Agent96]

Hi - a very quick question as I have been struggling with this for a while now. I have an Interface with a single method that takes a single parameter and has no return value. The parameter for this method is serializable but my implementation is wrappering an OutputStream - this is the key to this method, as the Object does all the work and writes the output directly to the stream - hiding the complexity from the client who uses this Object. Also, the fact it streams the data is key, as I do not want to hold all the information in memory.

Ideally I would like to call this method over a message channel using Spring Integration. I'm not sure this is even possible? I have experimented with different ideas, but am not sure what the best approach is to solve this problem - surely my use case is not unique, so I guess my question is are there any design patterns that solve this problem?

I guess one approach is to extend OutputStream with a Serializeable object whose implementation passes byte arrays back to the message channel...would this work/how would this work? Any ideas or suggestions? Or am I trying to do something totally weird and impossible....

[Gary Russell]

Why do you need to make it Serializable?

You can pass any arbitrary object (including an OutputStream) in a payload from one endpoint to another using Spring Integration.

Of course, it only works with in-memory channels; you can't store a live object such as an OutputStream in a database or JMS/AMQP queue.

Perhaps I am missing something in your question?

[Agent96]

Thanks for your reply. I think I must be missing something obvious. In my demos I am using a gateway, one with a service interface, and in the other demo with a service-activator, both have a response channel - in both cases they throw a NotSerializableException. Prehaps this is due to the response channel. In any case I'm clearly doing something stupid.

What I really need is an example of a client calling a method passing in an Object wrappering an output stream with the recieving method writing large volumes of data to the stream - more data then could fit into memory...or at least I need a prod in the right direction :-) Currently I just write the response to a file and pass back the file name for the calling client to read and copy to the ServletOutputStream, but I really don't want to do this, I'd prefer to pass back byte arrays in chunks...

[Gary Russell]

So you want to pass the ServletOutputStream in a message, and have something upstream write to it?

This is hardly loose coupling.

That said, I see no reason why it shouldn't work; you should only get serialization errors if you are sending your message through a persistent channel, or out to some other technology. Can you post a stack trace?

You might want to consider a different approach.

For example, from your servlet, call a <gateway/> that has a void return; hold up the thread to wait until the last packet is received.. Your downstream service can then send individual chunks (in byte arrays) in a wrapper containing correlation data. An outbound-channel adapter (hooked into your servlet) can receive these packets, use the correlation data to write them to the appropriate servlet output stream. When the last packet is received, wake up the container thread that sent the request.

[Agent96]

Thanks for your reply - the message is going to RabbitMQ. Here is the stack trace:

org.springframework.integration.MessageHandlingException: error occurred in message handler [org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint#0]
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:79)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)
	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)
	at org.springframework.integration.core.MessagingTemplate.doSendAndReceive(MessagingTemplate.java:318)
	at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:239)
	at org.springframework.integration.core.MessagingTemplate.convertSendAndReceive(MessagingTemplate.java:274)
	at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:224)
	at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceive(MessagingGatewaySupport.java:203)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.invokeGatewayMethod(GatewayProxyFactoryBean.java:305)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.doInvoke(GatewayProxyFactoryBean.java:268)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.invoke(GatewayProxyFactoryBean.java:259)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
	at $Proxy0.doSomethingGood(Unknown Source)
	at demo.metadatatechnology.manager.rabbit.DEMOInterface.Sender.main(Sender.java:25)
Caused by: org.springframework.amqp.support.converter.MessageConversionException: failed to convert to serialized Message content
	at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:141)
	at org.springframework.amqp.support.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:51)
	at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.sendAndReceive(AmqpOutboundEndpoint.java:154)
	at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.handleRequestMessage(AmqpOutboundEndpoint.java:129)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
	... 18 more
Caused by: java.lang.IllegalArgumentException: Could not serialize object of type: class demo.metadatatechnology.manager.rabbit.message.DemoObjectWithStream
	at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:47)
	at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:139)
	... 23 more
Caused by: java.io.NotSerializableException: org.apache.commons.io.output.ByteArrayOutputStream
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1156)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1474)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1392)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1150)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:326)
	at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:44)
	... 24 more
Exception in thread "main" org.springframework.amqp.support.converter.MessageConversionException: failed to convert to serialized Message content
	at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:141)
	at org.springframework.amqp.support.converter.AbstractMessageConverter.toMessage(AbstractMessageConverter.java:51)
	at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.sendAndReceive(AmqpOutboundEndpoint.java:154)
	at org.springframework.integration.amqp.outbound.AmqpOutboundEndpoint.handleRequestMessage(AmqpOutboundEndpoint.java:129)
	at org.springframework.integration.handler.AbstractReplyProducingMessageHandler.handleMessageInternal(AbstractReplyProducingMessageHandler.java:97)
	at org.springframework.integration.handler.AbstractMessageHandler.handleMessage(AbstractMessageHandler.java:73)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:115)
	at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:102)
	at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:77)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:157)
	at org.springframework.integration.channel.AbstractMessageChannel.send(AbstractMessageChannel.java:128)
	at org.springframework.integration.core.MessagingTemplate.doSend(MessagingTemplate.java:288)
	at org.springframework.integration.core.MessagingTemplate.doSendAndReceive(MessagingTemplate.java:318)
	at org.springframework.integration.core.MessagingTemplate.sendAndReceive(MessagingTemplate.java:239)
	at org.springframework.integration.core.MessagingTemplate.convertSendAndReceive(MessagingTemplate.java:274)
	at org.springframework.integration.gateway.MessagingGatewaySupport.doSendAndReceive(MessagingGatewaySupport.java:224)
	at org.springframework.integration.gateway.MessagingGatewaySupport.sendAndReceive(MessagingGatewaySupport.java:203)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.invokeGatewayMethod(GatewayProxyFactoryBean.java:305)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.doInvoke(GatewayProxyFactoryBean.java:268)
	at org.springframework.integration.gateway.GatewayProxyFactoryBean.invoke(GatewayProxyFactoryBean.java:259)
	at org.springframework.aop.framework.ReflectiveMethodInvocation.proceed(ReflectiveMethodInvocation.java:172)
	at org.springframework.aop.framework.JdkDynamicAopProxy.invoke(JdkDynamicAopProxy.java:202)
	at $Proxy0.doSomethingGood(Unknown Source)
	at demo.metadatatechnology.manager.rabbit.DEMOInterface.Sender.main(Sender.java:25)
Caused by: java.lang.IllegalArgumentException: Could not serialize object of type: class demo.metadatatechnology.manager.rabbit.message.DemoObjectWithStream
	at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:47)
	at org.springframework.amqp.support.converter.SimpleMessageConverter.createMessage(SimpleMessageConverter.java:139)
	... 23 more
Caused by: java.io.NotSerializableException: org.apache.commons.io.output.ByteArrayOutputStream
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1156)
	at java.io.ObjectOutputStream.defaultWriteFields(ObjectOutputStream.java:1509)
	at java.io.ObjectOutputStream.writeSerialData(ObjectOutputStream.java:1474)
	at java.io.ObjectOutputStream.writeOrdinaryObject(ObjectOutputStream.java:1392)
	at java.io.ObjectOutputStream.writeObject0(ObjectOutputStream.java:1150)
	at java.io.ObjectOutputStream.writeObject(ObjectOutputStream.java:326)
	at org.springframework.amqp.utils.SerializationUtils.serialize(SerializationUtils.java:44)
	... 24 more

But I think your second suggestion is actually a good approach - it is an something I have half coded up already while I was trying out ideas, so I'm pleased you've mentioned it as a good way forward. Thanks for your help.

[Gary Russell]

Yes, you can't send objects like OutputStreams out to AMQP.