Topic broadcast across dynamic cluster with core API

[frankgrimes97]

Hi All,

I have an application which embeds HornetQ and both produces and consumes using the in-vm transport.

I'd like to cluster the application so I have set up a cluster with dynamic discovery as outlined in the latest 2.3.x docs.

The message redistribution appears to be working correctly, so all looks good so far.

Now, I want to be able to send a cancel signal to any instance and have it broadcast the message to all participating nodes in the cluster.

I had thought to use a Topic but it appears that messages to it do not automatically propagate across the cluster.

From my understanding I need to set up a non-exclusive Divert configuration.

The problem is that I don't know all targets for the Divert ahead of time.

Is there any way to accomplish having the Divert configuration be dynamic as nodes join/leave the cluster?

Thanks,

Frank Grimes

[jbertram]

If I'm not mistaken, messages sent to a topic in a cluster will be sent to every node where there is a subscription.  Reference the "clustered-topic" example.  Is that not what you're seeing?

[clebert.suconic]

man.. I feel like a 5th grade kid trying to solve a math problem.. I didn't understand your question.. especially this part:

"Now, I want to be able to send a cancel signal to any instance and have it broadcast the message to all participating nodes in the cluster."

[frankgrimes97]

Heh, ok... I'll try to clarify what I'm trying to accomplish and what I'm seeing in my tests.

I have consumers which process clustered queue messages with message re-distribution enabled.

The contents of the queue messages are task IDs where the tasks (job requests) are stored in an external DB.

The processing can be time consuming and I need to be able to interrupt/cancel them.

I had thought to have a separate consumer thread in each of those clustered instances listening on Topics for cancel signals for particular task ID.

Any one of those instances could receive the signal and would post the cancel request in a way which would broadcast the message to all nodes participating in the cluster.

e.g.

Node1

  \- org.hornetq.api.core.client.ClientSession.createQueue("jobs", "jobs")

  \- org.hornetq.api.core.client.ClientSession.createQueue("cancel", "node1")

Node2

  \- org.hornetq.api.core.client.ClientSession.createQueue("jobs", "jobs")

  \- org.hornetq.api.core.client.ClientSession.createQueue("cancel", "node2")

Each node does a createProducer/createConsumer on "jobs".

Those messages are redistributed.

Each node does a createProducer/createConsumer on "cancel".

Those messages are only consumed by the node which produces the message.

I'd like every node to receive those messages.

I hope that helps explain things better.

Cheers,

Frank Grimes

[ataylor]

like justin said, just use a topic then all consumers in the cluster will get the message.

[frankgrimes97]

Hi Andy,

I thought I *was* using a Topic, as described here: http://docs.jboss.org/hornetq/2.3.0.Final/docs/user-manual/html/jms-core-mapping.html

Can someone post some sample code/pseudo-code showing how to use Topics with the core API?

Thanks,

Frank Grimes

[ataylor]

with the core API there is no such thing as topics we only deal with addresses and queues, a topic is basically an address with many queues (the subscription) where each queue has its own consumer as opposed to jms queues which is a single address with 1 queue and n consumers.

so a topic with 2 subscribers would be

address1 -> queue(sub1) -> consumer 1

              -> queue(sub2) -> consumer 2

[ataylor]

a bit more info

so to send a message to all the consumers you would send to address1 and then for each consumer create a queue with the same address but its own name and consume from this

[frankgrimes97]

Hi Andy,

From what I can tell, that's what I've done but I don't see those "Topic" messages being broadcast across the cluster.

If node1 posts to "cancel", only node1 receives that "cancel" message. Same for node2.

Not sure what I'm doing wrong...

Thanks,

Frank Grimes

[ataylor]

are you using unique queue names?

[frankgrimes97]

I believe so, yes.

node1

  \-- org.hornetq.api.core.client.ClientSession.createQueue("cancel", "node1")

  \-- org.hornetq.api.core.client.ClientSession.createProducer("cancel")

  \-- org.hornetq.api.core.client.ClientSession.createConsumer("node1")

node2

  \-- org.hornetq.api.core.client.ClientSession.createQueue("cancel", "node2")

  \-- org.hornetq.api.core.client.ClientSession.createProducer("cancel")

  \-- org.hornetq.api.core.client.ClientSession.createConsumer("node2")

[frankgrimes97]

I think found the problem.

I had neglected to create a ClusterConnectionConfiguration for this new "cancel" address.

So each Topic and/Or Queue needs its own ClusterConnectionConfiguration, correct?

[clebert.suconic]

you can have an expression covering all your queues and topics.

Since you used core.. you probably used an address name that wasn't picked up by your cluster connection.