Topics in ActiveMQ using JmsTemplate - ensuring delivery

[timothygordon32]

If you have a subscriber to a topic, then what properties should you be setting on the publishing JmsTemplate and/or the subscribing DefaultMessageListenerContainer if you'd like to ensure that no messages are missed by a subscriber and you'd like the subscriber to be tolerant of dropping out temporarily (ie gaps in network connectivity)?

When there are no subscribers, then using JConsole, I can see that is queuing messages, but the queue depth remains at zero. Ideally, I'd like the the messages to hang around a while so that subscribers have a chance to reconnect within a given time limit without missing any messages.

My apologies if this is more of a general/ActiveMQ question.

[denis.zhdanov]

...
My apologies if this is more of a general/ActiveMQ question.

This is mostly common JMS question. You should use durable topic consumers for ensuring that all received topic messages are processed. Check JMS specification for more details about that.

Also you can check persistent delivery mode and consider using queue instead of topic.

[timothygordon32]

I've investigated this further - durable topics are OK, but seem to me to be intended to handle the case where you'll have a well-managed lifecycle for the subscriber, like one of a set of client systems that may be intermittently available.

In my case I'm trying to handle transient clients on users workstations that come and go and there may be an unknown number of them at any one time.

I got some results, which I think are specific to ActiveMQ, using the consumer.retroactive=true property on the queue URI.

For those who are interested in the exact Spring configuration, here's the publisher config:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
    </bean>

    <bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="pubSubDomain" value="true"/>
        <property name="defaultDestinationName" value="TEST.TOPIC"/>
    </bean>

And here's the subscriber config:

    <bean id="connectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory">
        <property name="brokerURL" value="tcp://localhost:61616"/>
    </bean>

    <bean id="destination" class="org.apache.activemq.command.ActiveMQTopic">
        <property name="physicalName" value="TEST.TOPIC?consumer.retroactive=true&amp;consumer.prefetchSize=10"/>
    </bean>

    <bean id="jmsContainer" class="org.springframework.jms.listener.DefaultMessageListenerContainer">
        <property name="connectionFactory" ref="connectionFactory"/>
        <property name="destination" ref="destination"/>
        <property name="pubSubDomain" value="true"/>
        <property name="messageListener" ref="subscriber"/>
        <property name="messageSelector" value="username = '${username}'"/>
        <property name="recoveryInterval" value="60000"/>
    </bean>

(Note sure setting recoveryInterval did anything I could determine while I paused by subscribers in the debugger.)

I still don't think this addresses my original question of how I could publish a message to a topic that would live for say, an hour, and be picked up by any interested subscribers who happen to be around for that hour, and then be cleaned up by the queue manager.

[is_whchong]

Hi Tim,

Have you ever encounter problem of sending message to topics like below:

IllegalStateException: Session closed
	at com.ibm.ejs.jms.JMSSessionHandle.getOpenSession(JMSSessionHandle.java:1226)
	at com.ibm.ejs.jms.JMSSessionHandle.getOpenUnifiedSession(JMSSessionHandle.java:1263)
	at com.ibm.ejs.jms.JMSMessageProducerHandle.<init>(JMSMessageProducerHandle.java:129)
	at com.ibm.ejs.jms.JMSSessionHandle.createProducer(JMSSessionHandle.java:1777)
	at org.springframework.jms.core.JmsTemplate.doCreateProducer(JmsTemplate.java:846)
	at org.springframework.jms.core.JmsTemplate.createProducer(JmsTemplate.java:827)
	at org.springframework.jms.core.JmsTemplate.doSend(JmsTemplate.java:505)
	at org.springframework.jms.core.JmsTemplate$2.doInJms(JmsTemplate.java:477)
	at org.springframework.jms.core.JmsTemplate.execute(JmsTemplate.java:428)
	at org.springframework.jms.core.JmsTemplate.send(JmsTemplate.java:475)

[bsnyder]

I still don't think this addresses my original question of how I could publish a message to a topic that would live for say, an hour, and be picked up by any interested subscribers who happen to be around for that hour, and then be cleaned up by the queue manager.

If you need the message to live for an hour, then set the timeToLive property on the message for one hour. If the message is not consumed within an hour, the broker will discard it.

Bruce