STOMP message producer

[diktatoren]

Hi,

I'm trying to do the opposite of what's done in the StompExample; I'm trying to produce a message via the core api, and consume it using a STOMP client.

I've been using Jeff Mesnil's code from  http://jmesnil.net/weblog/2010/01/14/using-stomp-with-hornetq/ to create a message server that responds to STOMP. So far I've been unsuccessfull in my attempts to use the core API to produce messages to the server.

I've noticed that both examples use JMS and not the core API. Is there any reason to use JMS if JMS compatability is not important?

When using the following code to produce messages:

Producer:

          ClientSessionFactory nettyFactory =  HornetQClient.createClientSessionFactory(

                                                  new TransportConfiguration(

                                                     InVMConnectorFactory.class.getName()));

          ClientSession session = nettyFactory.createSession();

          ClientProducer producer = session.createProducer("jms.queue.a");

          ClientMessage message = session.createMessage(true);

          message.getBodyBuffer().writeString("Halloen");

          producer.send(message);

            session.start();

I get the following error when subscribing via STOMP:

SEVERE: Failed to process message due to: java.lang.NullPointerException. Message: HornetQMessage[null]:PERSISTENT

Could you please give me some pointers to what I'm doing wrong here?

[ataylor]

can you supply the full stacktrace

[timfox]

When making a post, as well as a full stack tracem, you should also post the full test case that replicates the problem, this is currently missing your stomp consumer code.

Also you didn't mention which stomp client you were using etc.

[diktatoren]

Here's the stack trace:

SEVERE: Failed to process message due to: java.lang.NullPointerException. Message: HornetQMessage[null]:PERSISTENT

java.lang.NullPointerException

at java.util.concurrent.ConcurrentHashMap.put(ConcurrentHashMap.java:910)

at org.codehaus.stomp.jms.ProtocolConverter.addMessageToAck(ProtocolConverter.java:190)

at org.codehaus.stomp.jms.StompSubscription.onMessage(StompSubscription.java:89)

at org.hornetq.jms.client.JMSMessageListenerWrapper.onMessage(JMSMessageListenerWrapper.java:91)

at org.hornetq.core.client.impl.ClientConsumerImpl.callOnMessage(ClientConsumerImpl.java:774)

at org.hornetq.core.client.impl.ClientConsumerImpl.access$100(ClientConsumerImpl.java:46)

at org.hornetq.core.client.impl.ClientConsumerImpl$Runner.run(ClientConsumerImpl.java:892)

at org.hornetq.utils.OrderedExecutorFactory$OrderedExecutor$1.run(OrderedExecutorFactory.java:96)

at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1110)

at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:603)

at java.lang.Thread.run(Thread.java:717)

My test case is start the server code (which is just Mesnil's code, re-arranged)

:

/**

* Create HornetQ Server with a Stomp Connector.

*/

public class HornetQStompServer extends Thread {

  HornetQServer hornetqServer;

  JMSServerManager jmsServer;

  StompConnect stompConnect;

public HornetQStompServer() throws Exception {

// to keep things simple, we disable security

Configuration configuration = new ConfigurationImpl();

configuration.setSecurityEnabled(false);

// we add a In-VM acceptor to HornetQ as the server will be accessible

// outside using Stomp

configuration.getAcceptorConfigurations().add(

new TransportConfiguration(InVMAcceptorFactory.class.getName()));

// we add a Queue which will be available to Stomp under /queue/a

configuration.getQueueConfigurations().add(new CoreQueueConfiguration("jms.queue.a", "jms.queue.a", null, true));

// we create the HornetQ server using this config

hornetqServer = HornetQServers

.newHornetQServer(configuration);

// we also create a JMS server manager as Stomp is using the JMS API

jmsServer = new JMSServerManagerImpl(hornetqServer);

// starting the JMS server will also start the underneath HornetQ server

// We create directly a JMS ConnectionFactory which will be

// connected to the HornetQ server using In-VM connection

ConnectionFactory connectionFactory = HornetQJMSClient

.createConnectionFactory(new TransportConfiguration(

InVMConnectorFactory.class.getName()));

// We inject the connection factory in Stomp

stompConnect = new StompConnect(connectionFactory);

// and start it using default Stomp config

}

  public void start() {

    try {

      jmsServer.start();

      stompConnect.start();

    } catch (Exception e) {

      e.printStackTrace();

    }

  }

*****

Start the producer and start a STOMP consumer, using a perl script that uses the Net::Stomp::Client module or connect via telnet (telnet localhost 61613) and input the following:

SUBSCRIBE

destination:/queue/a

ack:client

^@

Both these consumers work when a message is put in the queue, using Mesnil's example commands over telnet, but results in a NPE when messages are put in the queue using the java code above.

[timfox]

org.codehaus.stomp.jms.ProtocolConverter.addMessageToAck(ProtocolConverter.java:190)

This code is not HornetQ code, it's from the codehaus stomp client code.

You'd need to look at that code to figure out why it's throwing the NPE, or ask the codehaus people.

Wild guess: Since the class is jms.ProtocolConverter it's probably expecting a JMS message, and looking for some JMS header which clearly won't exist if it's a core message.

[jmesnil]

PÃ¥l Evensen wrote:

I've noticed that both examples use JMS and not the core API. Is there any reason to use JMS if JMS compatability is not important?

The example is using StompConnect which requires a JMS Provider. This will not work if you are using HornetQ Core API as it is missing headers mandatory for JMS (e.g. JMSMessageID).

Next release of HornetQ will allow to send/receive from HornetQ Core API to STOMP & vice versa.

[diktatoren]

Thanks to the both of you for helpfull answers.

Jeff: Isn't interoperability between the core API and STOMP supported in the trunk either? The generated documentation says otherwise.

[diktatoren]

Never mind, I got STOMP to work through the Core API