Transport Listener and ActiveMq restart

Question

I develope JMS application and now I want to add feature that will support broker restart. I have dynamic topics and I should recreate them after my connection will be resumed. Also I should have a possibility to know when broker is down and when it come back. So I try to implement this feature using ActiveMQ failover protocol. I implement TransportListener and in the method "transportInterrupted" I call full disconnect like

  public void disconnect() throws JMSException {
    System.out.println("!!!!!!!DISCONNECTING!!!!!!!!");
    consumer.close();
    session.close();
    session = null;
    messageProducer.close();
    messageProducer = null;
    connection = null;
    connected = false;
    System.out.println("!!!!!!!DISCONNECTED!!!!!!!!");
}

After this my application hangs and it's like race conditions. If i close() only producer and set connection to null everything is OK, but if i try to close consumer it works only in 1 from N cases. I write test that proves it. I think the problem in closing consumers, but i didn't find any information what i do wrong.

import javax.jms.Connection;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Queue;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;
public class FastFailProducer {
    volatile boolean connected = false;
    private ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory("failover:(tcp://localhost:61616)?timeout=5000");;
    private static FailoverListener failoverListener;
    private Connection connection;
    private Session session;
    private Queue queue;
    private MessageProducer messageProducer;
    private MessageConsumer consumer;
    private String something;

public void init() throws JMSException {
    System.out.println("!!!!!!!CONNECTING!!!!!!!!");
    connection = factory.createConnection();
    session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
    connection.start();
    ((ActiveMQConnection) connection).addTransportListener(failoverListener);
    queue = session.createQueue("TEST");
    messageProducer = session.createProducer(queue);
    consumer = session.createConsumer(queue);
    System.out.println("!!!!!!!CONNECTING COMPLETE!!!!!!!!");
    connected = true;
}

public void disconnect() throws JMSException {
    System.out.println("!!!!!!!DISCONNECTING!!!!!!!!");
    consumer.close();
    session.close();
    session = null;
    messageProducer.close();
    messageProducer = null;
    connection = null;
    connected = false;
    System.out.println("!!!!!!!DISCONNECTED!!!!!!!!");
}

public void run() throws Exception {
    // send messages
    for (int i = 0; i < 1000; i++) {
        if (connected) {
            if (session != null & messageProducer != null & queue != null) {
                // send a message
                messageProducer.send(session.createTextMessage(i + " message"));
                System.out.println("Sent message " + i);
            }
        } else {
            // execute your backup logic
            System.out.println("Message " + i + " not sent");

        }
        Thread.sleep(1000);
    }

    messageProducer.close();
    session.close();
    connection.close();
    System.exit(0);
}

public static void main(String[] args) throws Exception {
    FastFailProducer failoverProducer = new FastFailProducer();
    failoverProducer.something = "asdfasdf";
    failoverListener = new FailoverListener(failoverProducer);
    failoverProducer.init();
    failoverProducer.setConnected(true);
    failoverProducer.run();

}

public boolean isConnected() {
    return connected;
}

public void setConnected(boolean connected) {
    this.connected = connected;
}
}

TransportListenerImpl class

import java.io.IOException;

import javax.jms.JMSException;

import org.apache.activemq.transport.TransportListener;

public class FailoverListener implements TransportListener {
    private FastFailProducer failProducer;

public FailoverListener(FastFailProducer failProducer) {
    super();
    this.failProducer = failProducer;
}

@Override
public void onCommand(Object arg0) {
}

@Override
public void onException(IOException arg0) {

}

@Override
public void transportInterupted() {
    try {
        failProducer.disconnect();
    } catch (JMSException e) {
        e.printStackTrace();
    }
}

@Override
public void transportResumed() {
    System.out.println("!!!!!!!TRANSPORT RESUMED!!!!!!!!");
    if (!failProducer.isConnected()) {
        try {
            failProducer.init();
        } catch (JMSException e) {
            e.printStackTrace();
        }
    }
}
}

Tim Bish Tim Bish · Accepted Answer · 2013-01-05T17:53:04

I think you are missing the point of using the failover protocol. If you use failover then there's no need to close down the connection and its associated resources as the failover transport will take care of restoring everything on the client end as it was before the broker went down. Closing the connection in the event method will definitely lock as you are not expected to do such a thing. If you want to shut everything down when the broker goes away, don't use failover, and instead listen on the JMS exception listener event hook instead.

Transport Listener and ActiveMq restart

1 Answers