diff options
Diffstat (limited to 'sca-java-1.x/tags/1.6.1-TUSCANY-3909/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSMessageReceiver.java')
| -rw-r--r-- | sca-java-1.x/tags/1.6.1-TUSCANY-3909/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSMessageReceiver.java | 270 |
1 files changed, 270 insertions, 0 deletions
diff --git a/sca-java-1.x/tags/1.6.1-TUSCANY-3909/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSMessageReceiver.java b/sca-java-1.x/tags/1.6.1-TUSCANY-3909/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSMessageReceiver.java new file mode 100644 index 0000000000..e9e9f04ab2 --- /dev/null +++ b/sca-java-1.x/tags/1.6.1-TUSCANY-3909/binding-ws-axis2/src/main/java/org/apache/tuscany/sca/binding/ws/axis2/jms/JMSMessageReceiver.java @@ -0,0 +1,270 @@ +/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.tuscany.sca.binding.ws.axis2.jms;
+
+import java.io.InputStream;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Topic;
+import javax.naming.Context;
+import javax.xml.stream.XMLStreamException;
+
+import org.apache.axis2.AxisFault;
+import org.apache.axis2.Constants;
+import org.apache.axis2.addressing.RelatesTo;
+import org.apache.axis2.context.ConfigurationContext;
+import org.apache.axis2.context.MessageContext;
+import org.apache.axis2.description.Parameter;
+import org.apache.axis2.engine.AxisEngine;
+import org.apache.axis2.transport.jms.JMSConstants;
+import org.apache.axis2.transport.jms.JMSUtils;
+import org.apache.axis2.util.MessageContextBuilder;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import edu.emory.mathcs.backport.java.util.concurrent.Executor;
+
+/**
+ * This is the actual receiver which listens for and accepts JMS messages, and
+ * hands them over to be processed by a worker thread. An instance of this
+ * class is created for each JMSConnectionFactory, but all instances may and
+ * will share the same worker thread pool.
+ */
+public class JMSMessageReceiver implements MessageListener {
+
+ private static final Log log = LogFactory.getLog(JMSMessageReceiver.class);
+
+ /**
+ * The thread pool of workers
+ */
+ private Executor workerPool = null;
+ /**
+ * The Axis configuration context
+ */
+ private ConfigurationContext axisConf = null;
+ /**
+ * A reference to the JMS Connection Factory
+ */
+ private JMSConnectionFactory jmsConFac = null;
+
+ /**
+ * Create a new JMSMessage receiver
+ *
+ * @param jmsConFac the JMS connection factory associated with
+ * @param workerPool the worker thead pool to be used
+ * @param axisConf the Axis2 configuration
+ */
+ JMSMessageReceiver(JMSConnectionFactory jmsConFac,
+ Executor workerPool, ConfigurationContext axisConf) {
+ this.jmsConFac = jmsConFac;
+ this.workerPool = workerPool;
+ this.axisConf = axisConf;
+ }
+
+ /**
+ * Return the Axis configuration
+ *
+ * @return the Axis configuration
+ */
+ public ConfigurationContext getAxisConf() {
+ return axisConf;
+ }
+
+ /**
+ * Set the worker thread pool
+ *
+ * @param workerPool the worker thead pool
+ */
+ public void setWorkerPool(Executor workerPool) {
+ this.workerPool = workerPool;
+ }
+
+ /**
+ * The entry point on the recepit of each JMS message
+ *
+ * @param message the JMS message received
+ */
+ public void onMessage(Message message) {
+ // directly create a new worker and delegate processing
+ try {
+ if (log.isDebugEnabled()) {
+ StringBuffer sb = new StringBuffer();
+ sb.append("Received JMS message to destination : " + message.getJMSDestination());
+ sb.append("\nMessage ID : " + message.getJMSMessageID());
+ sb.append("\nCorrelation ID : " + message.getJMSCorrelationID());
+ sb.append("\nReplyTo ID : " + message.getJMSReplyTo());
+ log.debug(sb.toString());
+ }
+ } catch (JMSException e) {
+ e.printStackTrace();
+ }
+ workerPool.execute(new Worker(message));
+ }
+
+ /**
+ * Creates an Axis MessageContext for the received JMS message and
+ * sets up the transports and various properties
+ *
+ * @param message the JMS message
+ * @return the Axis MessageContext
+ */
+ private MessageContext createMessageContext(Message message) {
+
+ InputStream in = JMSUtils.getInputStream(message);
+
+ try {
+ MessageContext msgContext = axisConf.createMessageContext();
+
+ // get destination and create correct EPR
+ Destination dest = message.getJMSDestination();
+ String destinationName = null;
+ if (dest instanceof Queue) {
+ destinationName = ((Queue) dest).getQueueName();
+ } else if (dest instanceof Topic) {
+ destinationName = ((Topic) dest).getTopicName();
+ }
+
+ String serviceName = jmsConFac.getServiceByDestination(destinationName);
+
+ // hack to get around the crazy Active MQ dynamic queue and topic issues
+ if (serviceName == null) {
+ String provider = (String) jmsConFac.getProperties().get(
+ Context.INITIAL_CONTEXT_FACTORY);
+ if (provider.indexOf("activemq") != -1) {
+ serviceName = jmsConFac.getServiceNameForDestination(
+ ((dest instanceof Queue ?
+ JMSConstants.ACTIVEMQ_DYNAMIC_QUEUE :
+ JMSConstants.ACTIVEMQ_DYNAMIC_TOPIC) + destinationName));
+ }
+ }
+
+
+ if (serviceName != null) {
+ // set to bypass dispatching and handover directly to this service
+ msgContext.setAxisService(
+ axisConf.getAxisConfiguration().getService(serviceName));
+ }
+
+ msgContext.setIncomingTransportName(Constants.TRANSPORT_JMS);
+ msgContext.setTransportIn(
+ axisConf.getAxisConfiguration().getTransportIn(Constants.TRANSPORT_JMS));
+
+ msgContext.setTransportOut(
+ axisConf.getAxisConfiguration().getTransportOut(Constants.TRANSPORT_JMS));
+ // the reply is assumed to be on the JMSReplyTo destination, using
+ // the same incoming connection factory
+
+
+ JMSOutTransportInfo jmsOutTransportInfo;
+
+ if ((jmsConFac.getJndiUser() == null) || (jmsConFac.getJndiPass() == null))
+ jmsOutTransportInfo= new JMSOutTransportInfo(jmsConFac.getConFactory(), message.getJMSReplyTo());
+ else
+ jmsOutTransportInfo= new JMSOutTransportInfo(jmsConFac.getConFactory(), jmsConFac.getUser(), jmsConFac.getPass(), message.getJMSReplyTo());
+
+ msgContext.setProperty(Constants.OUT_TRANSPORT_INFO, jmsOutTransportInfo);
+
+ msgContext.setServerSide(true);
+ msgContext.setMessageID(message.getJMSMessageID());
+
+ Destination replyTo = message.getJMSReplyTo();
+ String jndiDestinationName = null;
+ if (replyTo == null) {
+ Parameter param = msgContext.getAxisService().getParameter(JMSConstants.REPLY_PARAM);
+ if (param != null && param.getValue() != null) {
+ jndiDestinationName = (String) param.getValue();
+ }
+ }
+
+ if (jndiDestinationName != null) {
+ msgContext.setReplyTo(jmsConFac.getEPRForDestination(jndiDestinationName));
+ }
+
+ String soapAction = JMSUtils.getProperty(message, JMSConstants.SOAPACTION);
+ if (soapAction != null) {
+ msgContext.setSoapAction(soapAction);
+ }
+
+ msgContext.setEnvelope(
+ JMSUtils.getSOAPEnvelope(message, msgContext, in));
+
+ // set correlation id
+ String correlationId = message.getJMSCorrelationID();
+ if (correlationId != null && correlationId.length() > 0) {
+ msgContext.setProperty(JMSConstants.JMS_COORELATION_ID, correlationId);
+ msgContext.setRelationships(
+ new RelatesTo[] { new RelatesTo(correlationId) });
+ }
+
+ return msgContext;
+
+ } catch (JMSException e) {
+ handleException("JMS Exception reading the destination name", e);
+ } catch (AxisFault e) {
+ handleException("Axis fault creating the MessageContext", e);
+ } catch (XMLStreamException e) {
+ handleException("Error reading the SOAP envelope", e);
+ }
+ return null;
+ }
+
+ private void handleException(String msg, Exception e) {
+ log.error(msg, e);
+ throw new AxisJMSException(msg, e);
+ }
+
+ /**
+ * The actual Runnable Worker implementation which will process the
+ * received JMS messages in the worker thread pool
+ */
+ class Worker implements Runnable {
+
+ private Message message = null;
+
+ Worker(Message message) {
+ this.message = message;
+ }
+
+ public void run() {
+ MessageContext msgCtx = createMessageContext(message);
+
+ AxisEngine engine = new AxisEngine(msgCtx.getConfigurationContext());
+ try {
+ log.debug("Delegating JMS message for processing to the Axis engine");
+ try {
+ engine.receive(msgCtx);
+ } catch (AxisFault e) {
+ log.debug("Exception occured when receiving the SOAP message", e);
+ if (msgCtx.isServerSide()) {
+ MessageContext faultContext = MessageContextBuilder.createFaultMessageContext(msgCtx, e);
+ engine.sendFault(faultContext);
+ }
+ }
+ } catch (AxisFault af) {
+ log.error("JMS Worker [" + Thread.currentThread().getName() +
+ "] Encountered an Axis Fault : " + af.getMessage(), af);
+ }
+ }
+ }
+}
|