| Methods Summary |
|---|
| protected void | doSend(javax.jms.Message msg, int deliveryMode, int priority, long timeToLive)Do the message send
QueueSession session = null;
try
{
session = connection.createQueueSession(false, QueueSession.AUTO_ACKNOWLEDGE);
QueueSender sender = session.createSender(dlq);
sender.send(msg, deliveryMode, priority, timeToLive);
}
catch (Throwable t)
{
handleSendError(msg, t);
}
finally
{
if (session != null)
{
try
{
session.close();
}
catch (Throwable t)
{
log.trace("Ignored ", t);
}
}
}
|
| protected int | getDeliveryMode(javax.jms.Message msg)Get the delivery mode for the DLQ message
try
{
return msg.getJMSDeliveryMode();
}
catch (Throwable t)
{
return Message.DEFAULT_DELIVERY_MODE;
}
|
| protected int | getPriority(javax.jms.Message msg)Get the priority for the DLQ message
try
{
return msg.getJMSPriority();
}
catch (Throwable t)
{
return Message.DEFAULT_PRIORITY;
}
|
| protected long | getTimeToLive(javax.jms.Message msg)Get the time to live for the DLQ message
try
{
long expires = msg.getJMSExpiration();
if (expires == Message.DEFAULT_TIME_TO_LIVE)
return Message.DEFAULT_TIME_TO_LIVE;
return expires - System.currentTimeMillis();
}
catch (Throwable t)
{
return Message.DEFAULT_TIME_TO_LIVE;
}
|
| protected abstract boolean | handleDelivery(javax.jms.Message msg)Do we handle the message?
|
| public boolean | handleRedeliveredMessage(javax.jms.Message msg)
boolean handled = handleDelivery(msg);
if (handled)
sendToDLQ(msg);
return handled;
|
| protected void | handleSendError(javax.jms.Message msg, java.lang.Throwable t)Handle a failure to send the message to the dlq
log.error("DLQ " + dlq + " error sending message " + msg, t);
|
| protected javax.jms.Message | makeWritable(javax.jms.Message msg)Make a writable copy of the message
boolean trace = log.isTraceEnabled();
try
{
HashMap tmp = new HashMap();
// Save properties
for (Enumeration en = msg.getPropertyNames(); en.hasMoreElements();)
{
String key = (String) en.nextElement();
tmp.put(key, msg.getObjectProperty(key));
}
// Make them writable
msg.clearProperties();
for (Iterator i = tmp.keySet().iterator(); i.hasNext();)
{
String key = (String) i.next();
try
{
msg.setObjectProperty(key, tmp.get(key));
}
catch (JMSException ignored)
{
if (trace)
log.trace("Could not copy message property " + key, ignored);
}
}
msg.setStringProperty(JBOSS_ORIG_MESSAGEID, msg.getJMSMessageID());
Destination destination = msg.getJMSDestination();
if (destination != null)
msg.setStringProperty(JBOSS_ORIG_DESTINATION, destination.toString());
return msg;
}
catch (Throwable t)
{
log.error("Unable to make writable " + msg, t);
return null;
}
|
| public void | messageDelivered(javax.jms.Message msg)
|
| public void | onException(javax.jms.JMSException exception)
activation.handleFailure(exception);
|
| protected void | sendToDLQ(javax.jms.Message msg)Send the message to the dlq
int deliveryMode = getDeliveryMode(msg);
int priority = getPriority(msg);
long timeToLive = getTimeToLive(msg);
// If we get a negative time to live that means the message has expired
if (timeToLive < 0)
{
if (log.isTraceEnabled())
log.trace("Not sending the message to the DLQ, it has expired " + msg);
return;
}
Message copy = makeWritable(msg);
if (copy != null)
doSend(copy, deliveryMode, priority, timeToLive);
|
| public void | setup(org.jboss.resource.adapter.jms.inflow.JmsActivation activation, javax.naming.Context ctx)
this.activation = activation;
setupDLQDestination(ctx);
setupDLQConnection(ctx);
|
| protected void | setupDLQConnection(javax.naming.Context ctx)Setup the DLQ Connection
JmsActivationSpec spec = activation.getActivationSpec();
String user = spec.getDLQUser();
String pass = spec.getDLQPassword();
String clientID = spec.getDLQClientID();
JMSProviderAdapter adapter = activation.getProviderAdapter();
String queueFactoryRef = adapter.getQueueFactoryRef();
log.debug("Attempting to lookup dlq connection factory " + queueFactoryRef);
QueueConnectionFactory qcf = (QueueConnectionFactory) Util.lookup(ctx, queueFactoryRef, QueueConnectionFactory.class);
log.debug("Got dlq connection factory " + qcf + " from " + queueFactoryRef);
log.debug("Attempting to create queue connection with user " + user);
if (user != null)
connection = qcf.createQueueConnection(user, pass);
else
connection = qcf.createQueueConnection();
if (clientID != null)
connection.setClientID(clientID);
connection.setExceptionListener(this);
log.debug("Using queue connection " + connection);
|
| protected void | setupDLQDestination(javax.naming.Context ctx)Setup the DLQ Destination
String name = activation.getActivationSpec().getDLQJNDIName();
dlq = (Queue) Util.lookup(ctx, name, Queue.class);
|
| public void | teardown()
teardownDLQConnection();
teardownDLQDestination();
|
| protected void | teardownDLQConnection()Teardown the DLQ Connection
try
{
if (connection != null)
{
log.debug("Closing the " + connection);
connection.close();
}
}
catch (Throwable t)
{
log.debug("Error closing the connection " + connection, t);
}
|
| protected void | teardownDLQDestination()Teardown the DLQ Destination
|
| protected void | warnDLQ(javax.jms.Message msg, int count, int max)Warn that a message is being handled by the DLQ
log.warn("Message redelivered=" + count + " max=" + max + " sending it to the dlq " + msg);
|
