8 Replies Latest reply: May 8, 2012 3:40 PM by Clebert Suconic RSS

Standalone HorentQ Topic Subscriber Stops Reading

floppydisk Newbie

For reference, this started as this StackOverflow discussion and Clerbert asked me to move it over here. Long story short, here's what's happening.

 

I'm exporting a durable topic from JBOSS 7.1.1.Final and have several standalone clients that read the data off the topic. Things start wonderfully, all the clients read the data successfully for a time span of 2-5 hours and then they just stop. The server keeps pumping data onto the topic but the clients stop reading data--I ascertained this by looking at the server log file and didn't see any disconnnects logged on the server side and the clients would just record a log message saying "MESSAGE RECEIVED" at <timestamp> and nothing after that, they just stop working and I've got no idea why.

 

Code to create the params for the connection factory:

 

private TransportConfiguration getTC(String hostname) {
       
Map<String,Object> params = new HashMap<String, Object>();
       
params.put(TransportConstants.HOST_PROP_NAME, hostname);
       
params.put(TransportConstants.PORT_PROP_NAME, 5445);
       
TransportConfiguration tc = new TransportConfiguration(NettyConnectorFactory.class.getName(), params);
       
return tc;

   
}

   
private Topic createDestination(String destinationName) {
       
Topic topic = new HornetQTopic(destinationName);
       
return topic;

   
}

   
private HornetQConnectionFactory createCF(TransportConfiguration tc) {
       
HornetQConnectionFactory cf = HornetQJMSClient.createConnectionFactoryWithoutHA(JMSFactoryType .CF, tc);
       
return cf == null ? null : cf;

   
}

 

 

Create and start session

TransportConfiguration tc = this.getTC(this.hostname);
       
HornetQConnectionFactory cf = this.createCF(tc);
        cf
.setRetryInterval(4000);
        cf
.setReconnectAttempts(10);
        cf
.setConfirmationWindowSize(1000000);

       
Destination destination = this.createDestination(this.topicName);
        logger
.info("Starting Topic Connection");
       
try {
           
this.connection = cf.createConnection();

            connection
.start();
           
this.session = connection.createSession(transactional, ackMode);
           
MessageConsumer consumer = session.createConsumer(destination);
            consumer
.setMessageListener(this);

            logger
.info("Started topic connection");
       
} catch (Exception ex) {
            ex
.printStackTrace();
            logger
.error("EXCEPTION!");
       
}

 

Finally, to do work, in the main loop I have:

 

try {

                              msg = this.consumer.receive();

                              msg.acknowledge();

                    } catch (JMSException e) {

                              // TODO Auto-generated catch block

                              e.printStackTrace();

                    }

 


I'm at a loss as to why the thing hangs up though.