/*
|
* 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.jpdl.internal.activity;
|
|
import java.io.Serializable;
|
import java.util.List;
|
|
import javax.jms.Destination;
|
import javax.jms.MapMessage;
|
import javax.jms.Message;
|
import javax.jms.MessageProducer;
|
import javax.jms.ObjectMessage;
|
import javax.jms.Queue;
|
import javax.jms.QueueConnection;
|
import javax.jms.QueueConnectionFactory;
|
import javax.jms.QueueSender;
|
import javax.jms.QueueSession;
|
import javax.jms.Session;
|
import javax.jms.TextMessage;
|
import javax.jms.Topic;
|
import javax.jms.TopicConnection;
|
import javax.jms.TopicConnectionFactory;
|
import javax.jms.TopicPublisher;
|
import javax.jms.TopicSession;
|
import javax.jms.XAConnectionFactory;
|
import javax.jms.XAQueueConnection;
|
import javax.jms.XAQueueConnectionFactory;
|
import javax.jms.XAQueueSession;
|
import javax.jms.XATopicConnection;
|
import javax.jms.XATopicConnectionFactory;
|
import javax.jms.XATopicSession;
|
import javax.naming.InitialContext;
|
|
import org.jbpm.api.JbpmException;
|
import org.jbpm.api.model.OpenExecution;
|
import org.jbpm.internal.log.Log;
|
import org.jbpm.pvm.internal.script.ScriptManager;
|
import org.jbpm.pvm.internal.wire.Descriptor;
|
import org.jbpm.pvm.internal.wire.WireContext;
|
import org.jbpm.pvm.internal.wire.descriptor.MapDescriptor;
|
|
/**
|
* @author Koen Aers
|
* @author Tom Baeyens
|
*/
|
public class JmsActivity extends JpdlAutomaticActivity {
|
|
private static final Log log = Log.getLog(JmsActivity.class.getName());
|
|
private static final long serialVersionUID = 1L;
|
|
protected String type = null;
|
protected String textExpression = null;
|
protected String objectExpression = null;
|
protected MapDescriptor mapDescriptor = null;
|
protected String connectionFactoryName = null;
|
protected String destinationName = null;
|
protected boolean transacted = true;
|
protected int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
|
|
public void perform(OpenExecution execution) {
|
try {
|
InitialContext initialContext = new InitialContext();
|
|
Destination destination = (Destination) initialContext.lookup(destinationName);
|
Object connectionFactory = initialContext.lookup(connectionFactoryName);
|
|
if (connectionFactory instanceof XAConnectionFactory) {
|
log.debug("connection factory '"+connectionFactoryName+"' is a XAConnectionFactory: using xa jms apis");
|
if (destination instanceof Queue) {
|
log.debug("destination '"+destinationName+"' is a Queue: using xa queue jms apis");
|
XAQueueConnectionFactory xaQueueConnectionFactory = (XAQueueConnectionFactory) connectionFactory;
|
sendToQueueXA((Queue) destination, xaQueueConnectionFactory);
|
|
} else if (destination instanceof Topic) {
|
log.debug("destination '"+destinationName+"' is a Topic: using xa topic jms apis");
|
XATopicConnectionFactory xaTopicConnectionFactory = (XATopicConnectionFactory) connectionFactory;
|
sendToTopicXA((Topic) destination, xaTopicConnectionFactory);
|
|
} else {
|
throw new JbpmException("invalid destination type for '"+destinationName+"': "+destination.getClass().getName());
|
}
|
|
} else { // non-XA
|
log.debug("connection factory '"+connectionFactoryName+"' is a ConnectionFactory: using non-xa jms apis");
|
if (destination instanceof Queue) {
|
log.debug("destination '"+destinationName+"' is a Queue: using non-xa queue jms apis");
|
QueueConnectionFactory queueConnectionFactory = (QueueConnectionFactory) connectionFactory;
|
sendToQueue((Queue) destination, queueConnectionFactory);
|
|
} else if (destination instanceof Topic) {
|
log.debug("destination '"+destinationName+"' is a Topic: using non-xa topic jms apis");
|
TopicConnectionFactory topicConnectionFactory = (TopicConnectionFactory) connectionFactory;
|
sendToTopic((Topic) destination, topicConnectionFactory);
|
|
} else {
|
throw new JbpmException("invalid destination type for '"+destinationName+"': "+destination.getClass().getName());
|
}
|
}
|
|
} catch (RuntimeException e) {
|
log.error("couldn't send jms message: "+e.getMessage(), e);
|
throw e;
|
|
} catch (Exception e) {
|
log.error("couldn't send jms message: "+e.getMessage(), e);
|
throw new JbpmException("couldn't send jms message to queue"+e.getMessage(), e);
|
}
|
}
|
|
protected void sendToQueueXA(Queue queue, XAQueueConnectionFactory xaQueueConnectionFactory) throws Exception {
|
XAQueueConnection xaQueueConnection = null;
|
XAQueueSession xaQueueSession = null;
|
MessageProducer messageProducer = null;
|
|
try {
|
xaQueueConnection = xaQueueConnectionFactory.createXAQueueConnection();
|
xaQueueSession = xaQueueConnection.createXAQueueSession();
|
messageProducer = xaQueueSession.createProducer(queue);
|
Message message = createMessage(xaQueueSession);
|
messageProducer.send(message);
|
|
} finally {
|
try {
|
messageProducer.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
xaQueueSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
xaQueueConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
protected void sendToTopicXA(Topic topic, XATopicConnectionFactory xaTopicConnectionFactory) throws Exception {
|
XATopicConnection xaTopicConnection = null;
|
XATopicSession xaTopicSession = null;
|
MessageProducer messageProducer = null;
|
|
try {
|
xaTopicConnection = xaTopicConnectionFactory.createXATopicConnection();
|
xaTopicSession = xaTopicConnection.createXATopicSession();
|
messageProducer = xaTopicSession.createProducer(topic);
|
Message message = createMessage(xaTopicSession);
|
messageProducer.send(message);
|
|
} finally {
|
try {
|
messageProducer.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
xaTopicSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
xaTopicConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
protected void sendToQueue(Queue queue, QueueConnectionFactory queueConnectionFactory) throws Exception {
|
QueueConnection queueConnection = null;
|
QueueSession queueSession = null;
|
QueueSender queueSender = null;
|
|
try {
|
queueConnection = queueConnectionFactory.createQueueConnection();
|
queueSession = queueConnection.createQueueSession(transacted, acknowledgeMode);
|
queueSender = queueSession.createSender(queue);
|
Message message = createMessage(queueSession);
|
queueSender.send(message);
|
if (transacted) {
|
queueSession.commit();
|
}
|
|
} finally {
|
try {
|
queueSender.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
queueSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
queueConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
protected void sendToTopic(Topic topic, TopicConnectionFactory topicConnectionFactory) throws Exception {
|
TopicConnection topicConnection = null;
|
TopicSession topicSession = null;
|
TopicPublisher topicPublisher = null;
|
|
try {
|
topicConnection = topicConnectionFactory.createTopicConnection();
|
topicSession = topicConnection.createTopicSession(transacted, acknowledgeMode);
|
topicPublisher = topicSession.createPublisher(topic);
|
Message message = createMessage(topicSession);
|
topicPublisher.send(message);
|
if (transacted) {
|
topicSession.commit();
|
}
|
|
} finally {
|
try {
|
topicPublisher.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
topicSession.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
try {
|
topicConnection.close();
|
} catch (Exception e) {
|
e.printStackTrace();
|
}
|
}
|
}
|
|
protected Message createMessage(Session session) throws Exception {
|
if ("text".equals(type)) {
|
return createTextMessage(session);
|
} else if ("object".equals(type)) {
|
return createObjectMessage(session);
|
} else if ("map".equals(type)) {
|
return createMapMessage(session);
|
}
|
throw new JbpmException("no type configured in jms activity");
|
}
|
|
private MapMessage createMapMessage(Session session) throws Exception {
|
MapMessage result = session.createMapMessage();
|
if (mapDescriptor != null) {
|
List<Descriptor> keyDescriptors = mapDescriptor.getKeyDescriptors();
|
List<Descriptor> valueDescriptors = mapDescriptor.getValueDescriptors();
|
for (int i = 0; i < keyDescriptors.size(); i++) {
|
String key = (String) WireContext.create(keyDescriptors.get(i));
|
Object value = (String) WireContext.create(valueDescriptors.get(i));
|
result.setObject(key, value);
|
}
|
}
|
return result;
|
}
|
|
private TextMessage createTextMessage(Session session) throws Exception {
|
Object value = ScriptManager.getScriptManager().evaluateExpression(textExpression, null);
|
if (value!=null) {
|
return session.createTextMessage(value.toString());
|
}
|
throw new JbpmException("null value for expression '"+textExpression+"' in jms activity");
|
}
|
|
private ObjectMessage createObjectMessage(Session session) throws Exception {
|
Object object = ScriptManager.getScriptManager().evaluateExpression(objectExpression, null);
|
if ( (object!=null)
|
&& !(object instanceof Serializable)
|
) {
|
throw new JbpmException("can't send jms message: creation of object message expression '"+objectExpression+"' must be done with serializable: "+object);
|
}
|
return session.createObjectMessage((Serializable) object);
|
}
|
|
public void setType(String type) {
|
this.type = type;
|
}
|
public void setText(String text) {
|
this.textExpression = text;
|
}
|
public void setExpression(String expression) {
|
this.objectExpression = expression;
|
}
|
public void setMapDescriptor(MapDescriptor mapDescriptor) {
|
this.mapDescriptor = mapDescriptor;
|
}
|
public void setConnectionFactoryName(String connectionFactoryName) {
|
this.connectionFactoryName = connectionFactoryName;
|
}
|
public void setDestinationName(String destinationName) {
|
this.destinationName = destinationName;
|
}
|
public boolean isTransacted() {
|
return transacted;
|
}
|
public void setTransacted(boolean transacted) {
|
this.transacted = transacted;
|
}
|
public int getAcknowledgeMode() {
|
return acknowledgeMode;
|
}
|
public void setAcknowledgeMode(int acknowledgeMode) {
|
this.acknowledgeMode = acknowledgeMode;
|
}
|
}
|
