package org.jboss.mq;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedList;
import javax.jms.ConnectionMetaData;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.JMSSecurityException;
import javax.jms.Queue;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import org.jboss.logging.Logger;
import org.jboss.mq.il.ClientILService;
import org.jboss.mq.il.ServerIL;
import org.jboss.util.UnreachableStatementException;
import EDU.oswego.cs.dl.util.concurrent.ClockDaemon;
import EDU.oswego.cs.dl.util.concurrent.Semaphore;
import EDU.oswego.cs.dl.util.concurrent.ThreadFactory;
public abstract class Connection implements Serializable, javax.jms.Connection
{
private static final long serialVersionUID = 87938199839407082L;
private static ThreadGroup threadGroup = new ThreadGroup("JBossMQ Client Threads");
static Logger log = Logger.getLogger(Connection.class);
static boolean trace = log.isTraceEnabled();
static protected ClockDaemon clockDaemon = new ClockDaemon();
public HashMap destinationSubscriptions = new HashMap();
public HashMap subscriptions = new HashMap();
public boolean modeStop;
protected ServerIL serverIL;
protected String clientID;
protected ConnectionToken connectionToken;
protected ClientILService clientILService;
protected long pingPeriod = 1000 * 60;
protected boolean ponged = true;
Semaphore pingTaskSemaphore = new Semaphore(1);
Object pingTaskId;
protected volatile boolean closing = false;
private volatile boolean setClientIdAllowed = true;
HashSet createdSessions;
int subscriptionCounter = Integer.MIN_VALUE;
Object subCountLock = new Object();
boolean closed;
SpyXAResourceManager spyXAResourceManager;
GenericConnectionFactory genericConnectionFactory;
private int lastMessageID;
private ExceptionListener exceptionListener;
private Object elLock = new Object();
private Thread elThread;
private StringBuffer sb = new StringBuffer();
private char[] charStack = new char[22];
String sessionId;
protected HashSet temps = new HashSet();
static
{
log.debug("Setting the clockDaemon's thread factory");
clockDaemon.setThreadFactory(new ThreadFactory()
{
public Thread newThread(Runnable r)
{
Thread t = new Thread(getThreadGroup(), r, "Connection Monitor Thread");
t.setDaemon(true);
return t;
}
});
}
public static ThreadGroup getThreadGroup()
{
if (threadGroup.isDestroyed())
threadGroup = new ThreadGroup("JBossMQ Client Threads");
return threadGroup;
}
Connection(String userName, String password, GenericConnectionFactory genericConnectionFactory) throws JMSException
{
createdSessions = new HashSet();
connectionToken = null;
closed = false;
lastMessageID = 0;
modeStop = true;
if (trace)
log.trace("Connection Initializing userName=" + userName + " " + this);
this.genericConnectionFactory = genericConnectionFactory;
genericConnectionFactory.initialise(this);
if (trace)
log.trace("Getting the serverIL " + this);
serverIL = genericConnectionFactory.createServerIL();
if (trace)
log.trace("serverIL=" + serverIL + " " + this);
try
{
authenticate(userName, password);
if (userName != null)
askForAnID(userName, password);
startILService();
}
catch (Throwable t)
{
try
{
serverIL.connectionClosing(null);
}
catch (Throwable t2)
{
log.debug("Error closing the connection", t2);
}
SpyJMSException.rethrowAsJMSException("Failed to create connection", t);
}
try
{
if (trace)
log.trace("Creating XAResourceManager " + this);
spyXAResourceManager = new SpyXAResourceManager(this);
if (trace)
log.trace("Starting the ping thread " + this);
startPingThread();
if (trace)
log.trace("Connection establishment successful " + this);
}
catch (Throwable t)
{
try
{
serverIL.connectionClosing(connectionToken);
}
catch (Throwable t2)
{
log.debug("Error closing the connection", t2);
}
try
{
stopILService();
}
catch (Throwable t2)
{
log.debug("Error stopping the client IL", t2);
}
SpyJMSException.rethrowAsJMSException("Failed to create connection", t);
}
}
Connection(GenericConnectionFactory genericConnectionFactory) throws JMSException
{
this(null, null, genericConnectionFactory);
}
public ServerIL getServerIL()
{
return serverIL;
}
public void asynchClose()
{
}
public void asynchDeleteTemporaryDestination(SpyDestination dest)
{
if (trace)
log.trace("Deleting temporary destination " + dest);
try
{
deleteTemporaryDestination(dest);
}
catch (Throwable t)
{
asynchFailure("Error deleting temporary destination " + dest, t);
}
}
public void asynchDeliver(ReceiveRequest requests[])
{
if (closing)
return;
if (trace)
log.trace("Async deliver requests=" + Arrays.asList(requests) + " " + this);
try
{
for (int i = 0; i < requests.length; i++)
{
if (trace)
log.trace("Processing request=" + requests[i] + " " + this);
SpyConsumer consumer = (SpyConsumer) subscriptions.get(requests[i].subscriptionId);
requests[i].message.createAcknowledgementRequest(requests[i].subscriptionId.intValue());
if (consumer == null)
{
send(requests[i].message.getAcknowledgementRequest(false));
log.debug("WARNING: NACK issued due to non existent subscription " + requests[i].message.header.messageId);
continue;
}
if (trace)
log.trace("Delivering messageid=" + requests[i].message.header.messageId + " to consumer=" + consumer);
consumer.addMessage(requests[i].message);
}
}
catch (Throwable t)
{
asynchFailure("Error during async delivery", t);
}
}
public void asynchFailure(String reason, Throwable t)
{
if (trace)
log.trace("Notified of failure reason=" + reason + " " + this, t);
if (closing)
return;
JMSException excep = SpyJMSException.getAsJMSException(reason, t);
synchronized (elLock)
{
ExceptionListener el = exceptionListener;
if (el != null && elThread == null)
{
try
{
Runnable run = new ExceptionListenerRunnable(el, excep);
elThread = new Thread(getThreadGroup(), run, "ExceptionListener " + this);
elThread.setDaemon(false);
elThread.start();
}
catch (Throwable t1)
{
log.warn("Connection failure: ", excep);
log.warn("Unable to start exception listener thread: ", t1);
}
}
else if (elThread != null)
log.warn("Connection failure, already in the exception listener", excep);
else
log.warn("Connection failure, use javax.jms.Connection.setExceptionListener() to handle this error and reconnect", excep);
}
}
public void asynchPong(long serverTime)
{
if (trace)
log.trace("PONG serverTime=" + serverTime + " " + this);
ponged = true;
}
public void deleteTemporaryDestination(SpyDestination dest) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("DeleteDestination dest=" + dest + " " + this);
try
{
serverIL.deleteTemporaryDestination(connectionToken, dest);
synchronized (subscriptions)
{
destinationSubscriptions.remove(dest);
}
synchronized (temps)
{
temps.remove(dest);
}
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot delete the TemporaryDestination", t);
}
}
public void setClientID(String cID) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (clientID != null)
throw new IllegalStateException("The connection has already a clientID");
if (setClientIdAllowed == false)
throw new IllegalStateException("SetClientID was not called emediately after creation of connection");
if (trace)
log.trace("SetClientID clientID=" + clientID + " " + this);
try
{
serverIL.checkID(cID);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot connect to the JMSServer", t);
}
clientID = cID;
connectionToken.setClientID(clientID);
}
public String getClientID() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
return clientID;
}
public ExceptionListener getExceptionListener() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
checkClientID();
return exceptionListener;
}
public void setExceptionListener(ExceptionListener listener) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
checkClientID();
exceptionListener = listener;
}
public ConnectionMetaData getMetaData() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
checkClientID();
return new SpyConnectionMetaData();
}
public synchronized void close() throws JMSException
{
if (closed)
return;
if (trace)
log.trace("Closing connection " + this);
closing = true;
exceptionListener = null;
JMSException exception = null;
try
{
doStop();
}
catch (Throwable t)
{
exception = SpyJMSException.getAsJMSException("Error duing stop", t);
}
if (trace)
log.trace("Closing sessions " + this);
Object[] vect = null;
synchronized (createdSessions)
{
vect = createdSessions.toArray();
}
for (int i = 0; i < vect.length; i++)
{
try
{
((SpySession) vect[i]).close();
}
catch (Throwable t)
{
if (exception == null)
exception = SpyJMSException.getAsJMSException("Error closing sessions", t);
}
}
if (trace)
log.trace("Closed sessions " + this);
if (trace)
log.trace("Notifying the server of close " + this);
try
{
serverIL.connectionClosing(connectionToken);
}
catch (Throwable t)
{
if (exception == null)
exception = SpyJMSException.getAsJMSException("Cannot close properly the connection", t);
}
if (trace)
log.trace("Stopping ping thread " + this);
try
{
stopPingThread();
}
catch (Throwable t)
{
if (exception == null)
exception = SpyJMSException.getAsJMSException("Cannot stop the ping thread", t);
}
if (trace)
log.trace("Stopping the ClientIL service " + this);
try
{
stopILService();
}
catch (Throwable t)
{
if (exception == null)
exception = SpyJMSException.getAsJMSException("Cannot stop the client il service", t);
}
closed = true;
if (trace)
log.trace("Disconnected from server " + this);
if (exception != null)
throw exception;
}
public void start() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
checkClientID();
if (modeStop == false)
return;
modeStop = false;
if (trace)
log.trace("Starting connection " + this);
try
{
serverIL.setEnabled(connectionToken, true);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot enable the connection with the JMS server", t);
}
}
public void stop() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
checkClientID();
doStop();
}
public String toString()
{
StringBuffer buffer = new StringBuffer();
buffer.append("Connection@").append(System.identityHashCode(this));
buffer.append('[');
if (connectionToken != null)
buffer.append("token=").append(connectionToken);
else
buffer.append("clientID=").append(clientID);
if (closed)
buffer.append(" CLOSED");
else if (closing)
buffer.append(" CLOSING");
buffer.append(" rcvstate=");
if (modeStop)
buffer.append("STOPPED");
else
buffer.append("STARTED");
buffer.append(']');
return buffer.toString();
}
String getNewMessageID() throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
synchronized (sb)
{
sb.setLength(0);
sb.append(clientID);
sb.append('-');
long time = System.currentTimeMillis();
int count = 0;
do
{
charStack[count] = (char) ('0' + (time % 10));
time = time / 10;
++count;
}
while (time != 0);
--count;
for (; count >= 0; --count)
{
sb.append(charStack[count]);
}
++lastMessageID;
if (lastMessageID < 0)
{
lastMessageID = 0;
}
int id = lastMessageID;
count = 0;
do
{
charStack[count] = (char) ('0' + (id % 10));
id = id / 10;
++count;
}
while (id != 0);
--count;
for (; count >= 0; --count)
{
sb.append(charStack[count]);
}
return sb.toString();
}
}
void addConsumer(SpyConsumer consumer) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
Subscription req = consumer.getSubscription();
synchronized (subCountLock)
{
req.subscriptionId = subscriptionCounter++;
}
req.connectionToken = connectionToken;
if (trace)
log.trace("addConsumer sub=" + req);
try
{
synchronized (subscriptions)
{
subscriptions.put(new Integer(req.subscriptionId), consumer);
LinkedList ll = (LinkedList) destinationSubscriptions.get(req.destination);
if (ll == null)
{
ll = new LinkedList();
destinationSubscriptions.put(req.destination, ll);
}
ll.add(consumer);
}
serverIL.subscribe(connectionToken, req);
}
catch (JMSSecurityException ex)
{
removeConsumerInternal(consumer);
throw ex;
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot subscribe to this Destination: ", t);
}
}
SpyMessage[] browse(Queue queue, String selector) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("Browsing queue=" + queue + " selector=" + selector + " " + this);
try
{
return serverIL.browse(connectionToken, queue, selector);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot browse the Queue", t);
throw new UnreachableStatementException();
}
}
void pingServer(long clientTime) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
trace = log.isTraceEnabled();
if (trace)
log.trace("PING " + clientTime + " " + this);
try
{
serverIL.ping(connectionToken, clientTime);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot ping the JMS server", t);
}
}
SpyMessage receive(Subscription sub, long wait) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("Receive subscription=" + sub + " wait=" + wait);
try
{
SpyMessage message = serverIL.receive(connectionToken, sub.subscriptionId, wait);
if (message != null)
message.createAcknowledgementRequest(sub.subscriptionId);
return message;
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot receive ", t);
throw new UnreachableStatementException();
}
}
void removeConsumer(SpyConsumer consumer) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
Subscription req = consumer.getSubscription();
if (trace)
log.trace("removeConsumer req=" + req);
try
{
serverIL.unsubscribe(connectionToken, req.subscriptionId);
removeConsumerInternal(consumer);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot unsubscribe to this destination", t);
}
}
void sendToServer(SpyMessage mes) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("SendToServer message=" + mes.header.jmsMessageID + " " + this);
try
{
serverIL.addMessage(connectionToken, mes);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot send a message to the JMS server", t);
}
}
void sessionClosing(SpySession who)
{
if (trace)
log.trace("Closing session " + who);
synchronized (createdSessions)
{
createdSessions.remove(who);
}
}
void unsubscribe(DurableSubscriptionID id) throws JMSException
{
if (trace)
log.trace("Unsubscribe id=" + id + " " + this);
try
{
serverIL.destroySubscription(connectionToken, id);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot destroy durable subscription " + id, t);
}
}
void checkTemporary(Destination destination) throws JMSException
{
if (destination instanceof TemporaryQueue || destination instanceof TemporaryTopic)
{
synchronized (temps)
{
if (temps.contains(destination) == false)
throw new JMSException("Cannot create a consumer for a temporary destination from a different session. " + destination);
}
}
}
synchronized protected void checkClientID() throws JMSException
{
if (setClientIdAllowed == false)
return;
setClientIdAllowed = false;
if (trace)
log.trace("Checking clientID=" + clientID + " " + this);
if (clientID == null)
{
askForAnID(); if (clientID == null)
throw new JMSException("Could not get a clientID");
connectionToken.setClientID(clientID);
if (trace)
log.trace("ClientID established " + this);
}
}
protected void askForAnID() throws JMSException
{
if (trace)
log.trace("Ask for an id " + this);
try
{
if (clientID == null)
clientID = serverIL.getID();
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot get a client ID", t);
}
}
protected void askForAnID(String userName, String password) throws JMSException
{
if (trace)
log.trace("Ask for an id user=" + userName + " " + this);
try
{
String configuredClientID = serverIL.checkUser(userName, password);
if (configuredClientID != null)
clientID = configuredClientID;
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot get a client ID", t);
}
}
protected void authenticate(String userName, String password) throws JMSException
{
if (trace)
log.trace("Authenticating user " + userName + " " + this);
try
{
sessionId = serverIL.authenticate(userName, password);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot authenticate user", t);
}
}
protected void send(AcknowledgementRequest item) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("Acknowledge item=" + item + " " + this);
try
{
serverIL.acknowledge(connectionToken, item);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot acknowlege a message", t);
}
}
protected void send(TransactionRequest transaction) throws JMSException
{
if (closed)
throw new IllegalStateException("The connection is closed");
if (trace)
log.trace("Transact request=" + transaction + " " + this);
try
{
serverIL.transact(connectionToken, transaction);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot process a transaction", t);
}
}
protected void startILService() throws JMSException
{
if (trace)
log.trace("Starting the client il " + this);
try
{
clientILService = genericConnectionFactory.createClientILService(this);
clientILService.start();
if (trace)
log.trace("Using client id " + clientILService + " " + this);
connectionToken = new ConnectionToken(clientID, clientILService.getClientIL(), sessionId);
serverIL.setConnectionToken(connectionToken);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot start a the client IL service", t);
}
}
protected void stopILService() throws JMSException
{
try
{
clientILService.stop();
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot stop a the client IL service", t);
}
}
public void doStop() throws JMSException
{
if (modeStop)
return;
modeStop = true;
if (trace)
log.trace("Stopping connection " + this);
try
{
serverIL.setEnabled(connectionToken, false);
}
catch (Throwable t)
{
SpyJMSException.rethrowAsJMSException("Cannot disable the connection with the JMS server", t);
}
}
private void removeConsumerInternal(SpyConsumer consumer)
{
synchronized (subscriptions)
{
Subscription req = consumer.getSubscription();
subscriptions.remove(new Integer(req.subscriptionId));
LinkedList ll = (LinkedList) destinationSubscriptions.get(req.destination);
if (ll != null)
{
ll.remove(consumer);
if (ll.size() == 0)
{
destinationSubscriptions.remove(req.destination);
}
}
}
}
private void startPingThread()
{
if (pingPeriod == 0)
return;
pingTaskId = clockDaemon.executePeriodically(pingPeriod, new PingTask(), true);
}
private void stopPingThread()
{
if (pingPeriod == 0)
return;
ClockDaemon.cancel(pingTaskId);
try
{
pingTaskSemaphore.attempt(1000 * 10);
}
catch (InterruptedException e)
{
Thread.currentThread().interrupt();
}
}
class PingTask implements Runnable
{
public void run()
{
try
{
pingTaskSemaphore.acquire();
}
catch (InterruptedException e)
{
log.debug("Interrupted requesting ping semaphore");
return;
}
try
{
if (ponged == false)
{
throw new SpyJMSException("No pong received", new IOException("ping timeout."));
}
ponged = false;
pingServer(System.currentTimeMillis());
}
catch (Throwable t)
{
asynchFailure("Unexpected ping failure", t);
}
finally
{
pingTaskSemaphore.release();
}
}
}
class ExceptionListenerRunnable implements Runnable
{
ExceptionListener el;
JMSException excep;
public ExceptionListenerRunnable(ExceptionListener el, JMSException excep)
{
this.el = el;
this.excep = excep;
}
public void run()
{
try
{
synchronized (el)
{
el.onException(excep);
}
}
catch (Throwable t)
{
log.warn("Connection failure: ", excep);
log.warn("Exception listener ended abnormally: ", t);
}
synchronized (elLock)
{
elThread = null;
}
}
}
}