core and "topics" without JMS

[unficyp]

Hi,

i try to implement a simple scenario to send messages between an embedded server and client ( 2 different processes connected via netty transport)

The system uses core api only, without any JMS stuff, except for the Topic idea.

I need the message published to all clients listening to this queue.

My Server created a Queue named fsevents, a producer, fetches data from an external source and writes the data to the queue:

{code}

sf = serverLocator.createSessionFactory();

coreSession = sf.createSession(true,true,0);

coreSession.createQueue(topicName, topicName, true);

producer = coreSession.createProducer(topicName);

{/code}

The client connects to the queue and waits for messages matching a specific selector:

{code}

session.createTemporaryQueue(topicName, clientID);

messageConsumer = session.createConsumer(topicName,"CCAgent = '"+filter+"'");

msg = messageConsumer.receive(5000);

{/code}

queueName = fsevents

clientID depends on the given commandline parameter

This works as long as i only start 1 client (e,g. clientID client1)

client1 receives all messages.

When i start another client with clientID client2, the behaviour is the same as a normal queue.

message1 to client1, message2 to client2

I thought a topic is basically a createTemporaryQueue(address,queuename), so my clients use

createQueue("fsevents","client1") and createTemporaryQueue("fsevents","client2")  and then set a consumer to listen to the queue.

I'm sure i'm missing something.

Why does my clients get only every 2nd message ?

thanks,

gw

[jmesnil]

You are confusing the semantics of HornetQ core API

Have a look at http://www.slideshare.net/gavinhogan/hornet-q-andthewebjudcon2010-7381151 (esp. starting from slide 53).

basically, using HornetQ core API, you send to an *address* and receives from *queues*

it's the way you bind queues to address that defines the messaging model

coreSession.createQueue(topicName, topicName, true);

=> You have created a queue with the name topicName (2nd param) bound to the *address* topicName (1st param)

session.createTemporaryQueue(topicName, clientID);

=> You create a temp queue named clientID bound to the *address* topicName

messageConsumer = session.createConsumer(topicName,"CCAgent = '"+filter+"'");

=> You have create a consumer which will receive message from the *queue* named topicName. That's not what you want.

From your code, the 2 consumers receives messages from the queue named *topicName*, not from the temp queues you have created.

Have a look at the HornetQ javadoc API to avoid confusing address, queues: http://docs.jboss.org/hornetq/2.3.0.Alpha/docs/api/org/hornetq/api/core/client/ClientSession.html

To fix this:

1. There is no need to create a queue for the producer, it will send messages to an *address* fsevents

2. Create temp queues named "client1" & "client2" bound to the "fsevents" address

3. create consumers for the temp queues "client1" & "client2" and you're all set.

jeff

[unficyp]

Thanks Jeff, the slides made it very clear.

Changes my code and it's working now.

Thanks !