/*
|
* JBoss, Home of Professional Open Source
|
* Copyright 2005, JBoss Inc., and individual contributors as indicated
|
* by the @authors tag. See the copyright.txt in the distribution for a
|
* full listing of individual contributors.
|
*
|
* This is free software; you can redistribute it and/or modify it
|
* under the terms of the GNU Lesser General Public License as
|
* published by the Free Software Foundation; either version 2.1 of
|
* the License, or (at your option) any later version.
|
*
|
* This software is distributed in the hope that it will be useful,
|
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
|
* Lesser General Public License for more details.
|
*
|
* You should have received a copy of the GNU Lesser General Public
|
* License along with this software; if not, write to the Free
|
* Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA
|
* 02110-1301 USA, or see the FSF site: http://www.fsf.org.
|
*/
|
package org.jbpm.test;
|
|
import java.util.HashMap;
|
import java.util.Map;
|
|
import javax.jms.Message;
|
import javax.jms.MessageConsumer;
|
import javax.jms.Queue;
|
import javax.jms.QueueConnection;
|
import javax.jms.QueueConnectionFactory;
|
import javax.jms.QueueSession;
|
import javax.jms.XAQueueConnection;
|
import javax.jms.XAQueueConnectionFactory;
|
import javax.jms.XAQueueSession;
|
import javax.naming.InitialContext;
|
import javax.naming.NamingException;
|
|
import junit.framework.AssertionFailedError;
|
|
import org.jbpm.api.JbpmException;
|
|
import com.mockrunner.ejb.JNDIUtil;
|
import com.mockrunner.jms.JMSTestModule;
|
import com.mockrunner.mock.jms.JMSMockObjectFactory;
|
import com.mockrunner.mock.jms.MockQueue;
|
import com.mockrunner.mock.jms.MockQueueConnectionFactory;
|
import com.mockrunner.mock.jms.MockTopic;
|
|
|
/**
|
* @author Tom Baeyens
|
*/
|
public abstract class JmsExtensions {
|
|
static {
|
try {
|
new InitialContext().getEnvironment();
|
} catch (NamingException e) {
|
try {
|
JNDIUtil.initMockContextFactory();
|
} catch (Exception e2) {
|
e.printStackTrace();
|
throw new RuntimeException("coudn't initialize mock jndi: "+e2.getMessage(), e2);
|
}
|
}
|
}
|
|
static Map<String, JMSMockObjectFactory> jmsMockObjectFactories = new HashMap<String, JMSMockObjectFactory>();
|
static Map<String, JMSTestModule> jmsTestModules = new HashMap<String, JMSTestModule>();
|
|
public static void createQueue(String connectionFactoryJndiName, String queueJndiName) {
|
MockQueue queue = getJmsTestModule(connectionFactoryJndiName)
|
.getDestinationManager()
|
.createQueue(queueJndiName);
|
|
bindToJndi(queueJndiName, queue);
|
}
|
|
public static void removeQueue(String connectionFactoryJndiName, String queueJndiName) {
|
getJmsTestModule(connectionFactoryJndiName)
|
.getDestinationManager()
|
.removeQueue(queueJndiName);
|
|
unbindFromJndi(queueJndiName);
|
}
|
|
public static void createTopic(String connectionFactoryJndiName, String topicJndiName) {
|
MockTopic topic = getJmsTestModule(connectionFactoryJndiName)
|
.getDestinationManager()
|
.createTopic(topicJndiName);
|
|
bindToJndi(topicJndiName, topic);
|
}
|
|
public static void removeTopic(String connectionFactoryJndiName, String topicJndiName) {
|
getJmsTestModule(connectionFactoryJndiName)
|
.getDestinationManager()
|
.removeTopic(topicJndiName);
|
|
unbindFromJndi(topicJndiName);
|
}
|
|
protected static void bindToJndi(String jndiName, Object object) {
|
try {
|
new InitialContext().bind(jndiName, object);
|
} catch (Exception e) {
|
throw new JbpmException("couldn't bind object '"+object+"' to jndi name '"+jndiName+"': "+e.getMessage(), e);
|
}
|
}
|
|
protected static void unbindFromJndi(String jndiName) {
|
try {
|
new InitialContext().unbind(jndiName);
|
} catch (Exception e) {
|
throw new JbpmException("couldn't unbind object from jndi name '"+jndiName+"': "+e.getMessage(), e);
|
}
|
}
|
|
private static JMSTestModule getJmsTestModule(String connectionFactoryJndiName) {
|
JMSTestModule jmsTestModule = jmsTestModules.get(connectionFactoryJndiName);
|
if (jmsTestModule==null) {
|
JMSMockObjectFactory jmsMockObjectFactory = getMockObjectFactory(connectionFactoryJndiName);
|
jmsTestModule = new JMSTestModule(jmsMockObjectFactory);
|
jmsTestModules.put(connectionFactoryJndiName, jmsTestModule);
|
}
|
return jmsTestModule;
|
}
|
|
private static JMSMockObjectFactory getMockObjectFactory(String connectionFactoryJndiName) {
|
JMSMockObjectFactory jmsMockObjectFactory = jmsMockObjectFactories.get(connectionFactoryJndiName);
|
if (jmsMockObjectFactory==null) {
|
jmsMockObjectFactory = new JMSMockObjectFactory();
|
jmsMockObjectFactories.put(connectionFactoryJndiName, jmsMockObjectFactory);
|
MockQueueConnectionFactory mockQueueConnectionFactory = jmsMockObjectFactory.getMockQueueConnectionFactory();
|
try {
|
new InitialContext().bind(connectionFactoryJndiName, mockQueueConnectionFactory);
|
} catch (Exception e) {
|
throw new JbpmException("couldn't bind mock queue connection factory '"+connectionFactoryJndiName+"': "+e.getMessage(), e);
|
}
|
}
|
return jmsMockObjectFactory;
|
}
|
|
public static Object consumeMessageFromQueueXA(String connectionFactoryJndiName, String queueJndiName, long timeout) {
|
try {
|
InitialContext context = new InitialContext();
|
Queue queue = (Queue)context.lookup(queueJndiName);
|
XAQueueConnectionFactory xaQueueConnectionFactory = (XAQueueConnectionFactory) context.lookup(connectionFactoryJndiName);
|
|
XAQueueConnection xaQueueConnection = null;
|
XAQueueSession queueSession = null;
|
MessageConsumer messageConsumer = null;
|
try {
|
xaQueueConnection = xaQueueConnectionFactory.createXAQueueConnection();
|
xaQueueConnection.start();
|
queueSession = xaQueueConnection.createXAQueueSession();
|
messageConsumer = queueSession.createConsumer(queue);
|
Message message = messageConsumer.receive(timeout);
|
if (message==null) {
|
throw new AssertionFailedError("no message on queue "+queueJndiName);
|
}
|
return message;
|
|
} finally {
|
try {
|
messageConsumer.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
queueSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
xaQueueConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
} catch (RuntimeException e) {
|
throw e;
|
} catch (Exception e) {
|
throw new JbpmException("couldn't receive message from queue: "+e.getMessage(), e);
|
}
|
}
|
|
public static Message consumeMessageFromQueue(String connectionFactoryJndiName, String queueJndiName, long timeout, boolean transacted, int acknowledgeMode) {
|
try {
|
InitialContext context = new InitialContext();
|
QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) context.lookup(connectionFactoryJndiName);
|
Queue queue = (Queue)context.lookup(queueJndiName);
|
|
QueueConnection queueConnection = null;
|
QueueSession queueSession = null;
|
MessageConsumer messageConsumer = null;
|
|
try {
|
queueConnection = queueConnectionFactory.createQueueConnection();
|
queueConnection.start();
|
queueSession = queueConnection.createQueueSession(transacted, acknowledgeMode);
|
messageConsumer = queueSession.createConsumer(queue);
|
Message message = messageConsumer.receive(timeout);
|
if (message==null) {
|
throw new AssertionFailedError("no message on queue "+queueJndiName);
|
}
|
if (transacted) {
|
queueSession.commit();
|
}
|
return message;
|
|
} finally {
|
try {
|
messageConsumer.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
queueSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
queueConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
} catch (RuntimeException e) {
|
throw e;
|
} catch (Exception e) {
|
throw new JbpmException("couldn't receive message from queue: "+e.getMessage(), e);
|
}
|
}
|
|
public static void jmsAssertQueueEmptyXA(String connectionFactoryJndiName, String queueJndiName, long timeout) {
|
try {
|
consumeMessageFromQueueXA(connectionFactoryJndiName, queueJndiName, timeout);
|
} catch (AssertionFailedError e) {
|
if (e.getMessage().startsWith("no message on queue")) {
|
return;
|
}
|
}
|
throw new AssertionFailedError("message available on queue "+queueJndiName);
|
}
|
|
public static void jmsAssertQueueEmpty(String connectionFactoryJndiName, String queueJndiName, long timeout, boolean transacted, int acknowledgeMode) {
|
try {
|
consumeMessageFromQueue(connectionFactoryJndiName, queueJndiName, timeout, transacted, acknowledgeMode);
|
} catch (AssertionFailedError e) {
|
if (e.getMessage().startsWith("no message on queue")) {
|
return;
|
}
|
}
|
throw new AssertionFailedError("message available on queue "+queueJndiName);
|
}
|
}
|
