Invalid stream header: 61736466 when trying to read message published in management interface

[Eric B]

I'm trying to do a quick test with Spring, RabbitMQ and RabbitMQ JMS. I have configured my application with:

• Spring Boot 1.5.6
• RabbitMQ-jms 1.7.0
• RabbitMQ amqp-client 4.2.0
• RabbitMQ Server 3.6.11 (running in a docker container)

When I listen for a message in RabbitMQ, and publish it via the Management Interface, I get a StreamCorruptedException : invalid stream header: 61736466 thrown.

Digging through the code, I see this is thrown by the call in WhiteListObjectInputStream().super(). In the java.io.ObjectInputStream() constructor, it checks the stream header for the STREAM_MAGIC & STREAM_VERSION.

Is this because it was a non-Java library (ie: the management interface) that created the message? Is there a problem with the way I created the connection? Is there an issue with the amqp library?

@MichaelKlishin responded by indicating that "JMS client assumes it interoperates with either JMS client or (with certain restrictions) RabbitMQ Java client. Management UI publishing form sets an absolute minimum of message properties by default"

But would that be contrary to the idea?  Shouldn't I be able to publish messages into RabbitMQ using whatever technology stack I want, and have my Java consumer/subscriber be able to receive the message?  Does my producer/publisher HAVE to be Java based in order to use the JMS/Java client?

Thanks,

Eric

[Luke Bakken]

Hi Eric,

This may help explain what you're seeing: https://spring.io/understanding/AMQP

"A limitation of JMS is that the APIs are specified, but the message format is not"

Thanks,

Luke

[Eric B]

Thanks Luke, 

But I'm not sure I understand how that would be applicable; even if not defined by JMS, the message format would be defined by RabbitMQ (or AMQP).  It should then be used by all RabbitMQ clients - both the Java client and whatever the Management Interface uses.  In either case, I would expect that any client able to communicate with RabbitMQ would be compatible with any other client.

Thanks,

Eric

--

You received this message because you are subscribed to a topic in the Google Groups "rabbitmq-users" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/rabbitmq-users/Zbcp9WL-vUM/unsubscribe.
To unsubscribe from this group and all its topics, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitm...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Arnaud Cogoluègnes]

Hi, Eric.

I'm not sure to understand what you want to achieve, could you provide some basic code? (if possible, remove the Spring AMQP and Spring Boot parts, it's better to have less abstractions if we want to focus on JMS interoperability in RabbitMQ).

BTW, it's possible to consume with the JMS client messages from any kind of source (Java Client or not).

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.
To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

[Luke Bakken]

Hi Eric,

JMS standardizes an API, not a message format. It just so happens that AMQP 0-9-1 is used "behind the scenes" with additional, JMS-specific information, as you have found.

See the "JMS and AMQP 0-9-1" section of this document: https://www.rabbitmq.com/jms-client.html

If you need AMQP 0-9-1 interoperability, use the RabbitMQ AMQP Java client.

Thanks,

Luke

[Eric B]

Hi Arnaud,

My test case is very basic.  I have a Spring JMS listener bean:

@Component

public class Receiver {

    @JmsListener(destination = "mailbox")

    public void receiveMessage(Message msg) {

    log.info("Received <" + msg + ">");

    }

}

Where I've defined my jmsListenerContainerFactory and my connectionFactory as Spring Beans:

@Bean

public DefaultJmsListenerContainerFactory jmsListenerContainerFactory() {

DefaultJmsListenerContainerFactory factory = new DefaultJmsListenerContainerFactory();

factory.setConnectionFactory(connectionFactory());

factory.setDestinationResolver(new DynamicDestinationResolver());

return factory;

}

@Bean

public ConnectionFactory connectionFactory() {

RMQConnectionFactory conn = new RMQConnectionFactory();

conn.setUsername("kodo");

conn.setPassword("kodo");

conn.setHost("localhost");

conn.setPort(5672);

conn.setVirtualHost("/");

return conn;

}

However, as I indicated previously, if I try to post a message to the Queue via the Management Interface, I get the StreamCorruptedException.  Debugging pointed me to the client having trouble reading the stream, that there were no special MAGIC or VERSION in the stream, and hence the invalid stream header.  

If I try to post using a JMS client, it works fine.  But then that means that I need to us a JMS client (or Java?) as a publisher, and wouldn't be able to use this as a cross-platform system. 

I would have expected to be able to use any technology to post my message, and have the JMS client be able to read it.  While it may not have been able to interpret it (ie: might not know how to unserialize the data), it should at least be able to read the stream.  Here it is failing even reading the stream altogether.

To me, this sounds like a bug/gap in the JMS client, but I'm not sure if my expectations for interoperability are wrong.

Thanks,

Eric

To unsubscribe from this group and all its topics, send an email to rabbitmq-user...@googlegroups.com.

To post to this group, send email to rabbitm...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

--
You received this message because you are subscribed to the Google Groups "rabbitmq-users" group.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-user...@googlegroups.com.

[Eric B]

Hi Luke,

Does that then imply that using the JMS connector limits this to reading JMS based messages?  ie: I can only use a JMS publisher if I plan to subscribe using JMS?  That still sounds backwards to me; I always expected any JMS supported message broker to be compatible with all its clients - not just JMS based ones.

Thanks,

Eric

On Tue, Aug 29, 2017 at 11:49 AM, Luke Bakken <lba...@pivotal.io> wrote:

Hi Eric,

JMS standardizes an API, not a message format. It just so happens that AMQP 0-9-1 is used "behind the scenes" with additional, JMS-specific information, as you have found.

See the "JMS and AMQP 0-9-1" section of this document: https://www.rabbitmq.com/jms-client.html

If you need AMQP 0-9-1 interoperability, use the RabbitMQ AMQP Java client.

Thanks,

Luke

On Monday, August 28, 2017 at 8:52:20 PM UTC-7, Eric B wrote:

Thanks Luke, 

But I'm not sure I understand how that would be applicable; even if not defined by JMS, the message format would be defined by RabbitMQ (or AMQP).  It should then be used by all RabbitMQ clients - both the Java client and whatever the Management Interface uses.  In either case, I would expect that any client able to communicate with RabbitMQ would be compatible with any other client.

Thanks,

Eric

On Mon, Aug 28, 2017 at 10:54 PM, Luke Bakken <lba...@pivotal.io> wrote:

Hi Eric,

This may help explain what you're seeing: https://spring.io/understanding/AMQP

"A limitation of JMS is that the APIs are specified, but the message format is not"

Thanks,

Luke

--

You received this message because you are subscribed to a topic in the Google Groups "rabbitmq-users" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/rabbitmq-users/Zbcp9WL-vUM/unsubscribe.
To unsubscribe from this group and all its topics, send an email to rabbitmq-users+unsubscribe@googlegroups.com.

To post to this group, send email to rabbitmq-users@googlegroups.com.

[Luke Bakken]

Hi Eric, 

Yes, that's my understanding after doing some research. I will note that this line in the docs does seem to imply some interoperability.

"Because JMS Client for RabbitMQ is implemented using the RabbitMQ Java client, it is compliant with both the JMS API and the AMQP 0-9-1 protocol."

I'll let Arnaud comment on that and continue on this thread as he's the expert!

Thanks,

Luke

[Arnaud Cogoluègnes]

You may have to create your own Spring's DestinationResolver.

You should experiment with simple code like this:

RMQConnectionFactory connectionFactory = new RMQConnectionFactory();

connectionFactory.setUsername("guest");

connectionFactory.setPassword("guest");

connectionFactory.setVirtualHost("/");

connectionFactory.setHost("localhost");

RMQDestination jmsDestination = new RMQDestination();

jmsDestination.setAmqp(true);

jmsDestination.setAmqpQueueName("rabbitQueueName");

Connection connection = connectionFactory.createConnection();

connection.start();

Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

RMQMessageConsumer consumer = (RMQMessageConsumer) session.createConsumer(jmsDestination);

consumer.setMessageListener(new MessageListener() {

    @Override

    public void onMessage(Message message) {

        System.out.println(message.getClass());

        System.out.println(message);

    }

});

And publish to the rabbitQueueName queue (it needs to be created first) from the Management UI. If you publish text-based messages, you can set up a JMSType property to TextMessage in the Management UI.

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-users+unsubscribe@googlegroups.com.
To post to this group, send email to rabbitmq-users@googlegroups.com.

[Eric B]

Thanks for the snippet of code Arnaud - it helped a lot.  

I am confused about some parameters in the Destination however.  According to the docs (https://www.rabbitmq.com/jms-client.html), the Destination has a JNDI name attribute, as well as a destinationName property.

The docs indicate "Name of the JMS destination.".  What is that supposed to mean?  Which destination?  How is this destinationName used?  I'm sure I'm missing something obvious, but not sure what.   How does this differ from the JNDI name attribute?

Thanks,

Eric

To unsubscribe from this group and all its topics, send an email to rabbitmq-users+unsubscribe@googlegroups.com.

[Arnaud Cogoluègnes]

There's actually no "name" property in RMQDestination, it refers to the JNDI name in the container and is used only when declaring the destination with the container-specific syntax (usually a XML "name" attribute). That's why the "JNDI Only?" column says "Yes" for this field.

"destinationName" is used only in the context of JNDI (see Javadoc and [1]).

You should experiment more with simple snippets and don't hesitate to dive into the code.

[1] https://github.com/rabbitmq/rabbitmq-jms-client/blob/b5394b48a2e96c60e1c1d2a3c3d392dc47abd4c9/src/main/java/com/rabbitmq/jms/admin/RMQDestination.java#L264

[Павел Алехин]

I have the same problem.

Environment:

• Apache Tomcat 8.5.35
  
• Spring Framework 4.3.2
  
• RabbitMQ-jms 1.11.0
• RabbitMQ amqp-client 4.8.3
• RabbitMQ Server 3.7.8

When message published via JMS client, it's received without issues.

And when I try to publish in Management web-interface, jms-client raises the same Exception: invalid stream header.

Tomcat server.xml snippet:

<Resource factory="com.rabbitmq.jms.admin.RMQObjectFactory"
            host="localhost" name="jms/connectionFactory" password="guest"
            port="5672" type="javax.jms.ConnectionFactory" username="guest"
            virtualHost="/" />

Context.xml:

<ResourceLink name="jms/connectionFactory"
        global="jms/connectionFactory" type="javax.jms.ConnectionFactory" />

Spring-config.xml snippet:

<jee:jndi-lookup id="jmsConnection"
        jndi-name="jms/connectionFactory" />

    <bean id="jmsDestinationResolver"
        class="org.springframework.jms.support.destination.DynamicDestinationResolver" />

    <bean id="debugLogger"
        class="integration.AppJmsMessagingTemplate"
        p:connectionFactory-ref="jmsConnection"
        p:destinationResolver-ref="jmsDestinationResolver"
        p:defaultDestinationName="logs.topic" p:pubSubDomain="true" />

    <bean id="jmsLogListener"
        class="integration.JmsLogListener" />

    <bean id="jmsListenerContainer"
        class="org.springframework.jms.listener.DefaultMessageListenerContainer"
        p:connectionFactory-ref="jmsConnection"
        p:destinationName="logs.topic"
        p:destinationResolver-ref="jmsDestinationResolver"
        p:messageListener-ref="jmsLogListener" p:pubSubDomain="true" />

When the spring-context initialized, exchange and queue had been created automatically by rabbitmq-jms-client, I suppose.

jms.durable.topic

queue: jms-cons-8946a69e-8ad9-4e60-a294-bb35a3ec9ee0 with routing key logs.topic

Btw, why simple topic exchange was created? Not a x-jms-topic?

So, I have the same problem, as Eric.

Publishing from jms-client is OK, but via management interface raises an exception.

The issue is in RMQMessage.fromMessage method.

Exception is raised on WhiteListObjectInputStream creating.

WhiteListObjectInputStream extends ObjectInputStream.

ObjectInputStream on creating checks StreamHeader's first two bytes to be STREAM_MAGIC and STREAM_VERSION consts.

If not, StreamCorruptedException is raised.

So, when jms-client sends a mesasge, it's serialized as an ObjectOutputStream, and receiving is OK.

Management interface sends just payload as a byte-array without control bytes, and it's raising an exception.

Finally, have the same question, as Eric: Does JMS client assume, it interoperates with either JMS client?

If it does, it's not a cross-platform solution, and it's unable to receive jms-messages, published by amqp-clients to the same destination.

Is it a bug?

May be, we should do some extra config?

Has anyone found a solution?

четверг, 31 августа 2017 г., 16:57:57 UTC+3 пользователь Arnaud Cogoluègnes написал:

To unsubscribe from this group and stop receiving emails from it, send an email to rabbitmq-user...@googlegroups.com.

To post to this group, send email to rabbitm...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

--
You received this message because you are subscribed to a topic in the Google Groups "rabbitmq-users" group.
To unsubscribe from this topic, visit https://groups.google.com/d/topic/rabbitmq-users/Zbcp9WL-vUM/unsubscribe.

To unsubscribe from this group and all its topics, send an email to rabbitmq-user...@googlegroups.com.
To post to this group, send email to rabbitm...@googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[Arnaud Cogoluègnes]

Have you tried to set the "JMSType" header to "TextMessage" in the
management UI? This way the driver will convert the AMQP message into
a JMS TextMessage.

[Arnaud Cogoluègnes]

The JMS client provides interoperability with "JMS AMQP" destinations
[1], and this is where the "JMSType" header can come up handy.
Unfortunately JMS AMQP destinations are not supported for JMS topics
and this is what you're using. So there's no interoperability here,
I'll update the documentation to mention this. You can scratch what
I've said in my previous email.

Regarding your question "why simple topic exchange was created? Not a
x-jms-topic?", this is because you didn't use a selector. An
x-jms-topic is placed automatically between the first exchange and
consumer AMQP queue to enforce the selector when one is specified
during the JMS topic subscriber is created.

[1] http://www.rabbitmq.com/jms-client.html#destination-interoperability

[Павел Алехин]

It doesn't help.

Tried in the management UI and also with the rabbitmqadmin via cmd-line:

python.exe rabbitmqadmin publish exchange=jms.durable.topic routing_key=logs.topic payload="hello, world" properties={\"headers\":{\"JMSType\":\"TextMessage\"}}

The issue is the same: invalid stream header.

вторник, 20 ноября 2018 г., 20:12:32 UTC+3 пользователь Arnaud Cogoluègnes написал:

[Arnaud Cogoluègnes]

Yes, you need to use a JMS queue and to use a JMS AMQP destination
(see my previous email for more details).

[Павел Алехин]

Tried directly with queue. It works fine.

But we need a publish-subscribe mode.

So we have to use RabbitMQ JMS Client only.

Thanks a lot for help!

среда, 21 ноября 2018 г., 13:35:10 UTC+3 пользователь Arnaud Cogoluègnes написал:

[Arnaud Cogoluègnes]

What is your use case exactly? I understood the JMS subscribers should
be able to consume messages sent by AMQP clients, right? What kind of
messages that would be in the end (TextMessage, ByteMessages)?

[Павел Алехин]

It's just a logging system.

All possible publishers and subscribers are java applications right now.

So RabbitMQ JMS Client is enough for us.

And it was just some research, what we can do, if new  AMQP clients would send messages in future.

среда, 21 ноября 2018 г., 19:03:18 UTC+3 пользователь Arnaud Cogoluègnes написал:

[Arnaud Cogoluègnes]

JMS topic subscribers being able to consume messages from non-JMS
publishers seem a reasonable feature to me, so I created a follow-up
issue [1].

[1] https://github.com/rabbitmq/rabbitmq-jms-client/issues/71