summaryrefslogtreecommitdiffstats
path: root/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider
diff options
context:
space:
mode:
Diffstat (limited to 'sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider')
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java117
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingInvoker.java273
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingListener.java169
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java62
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java142
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java227
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java55
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java176
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java70
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java69
-rw-r--r--sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java91
11 files changed, 1451 insertions, 0 deletions
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java
new file mode 100644
index 0000000000..0b7de9f401
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java
@@ -0,0 +1,117 @@
+/*
+ * 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.jms.provider;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.Session;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.osoa.sca.ServiceRuntimeException;
+
+/**
+ * Base MessageProcessor for the JMSBinding.
+ *
+ * @version $Rev$ $Date$
+ */
+public abstract class AbstractMessageProcessor implements JMSMessageProcessor {
+
+ protected String operationPropertyName;
+ protected boolean xmlFormat = true;
+
+ public AbstractMessageProcessor(JMSBinding jmsBinding) {
+ this.operationPropertyName = jmsBinding.getOperationSelectorPropertyName();
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.OperationAndDataBinding#getOperationName(javax.jms.Message)
+ */
+ public String getOperationName(Message message) {
+ try {
+
+ return message.getStringProperty(operationPropertyName);
+
+ } catch (JMSException e) {
+ throw new JMSBindingException("Exception retreiving operation name from message", e);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.OperationAndDataBinding#setOperationName(javax.jms.Message, java.lang.String)
+ */
+ public void setOperationName(String operationName, Message message) {
+ try {
+
+ message.setStringProperty(operationPropertyName, operationName);
+
+ } catch (JMSException e) {
+ throw new JMSBindingException("Exception setting the operation name on message", e);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.OperationAndDataBinding#extractPayload(javax.jms.Session, java.lang.Object)
+ */
+ public Message insertPayloadIntoJMSMessage(Session session, Object o) {
+ return createJMSMessage(session, o);
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.OperationAndDataBinding#extractPayload(javax.jms.Message)
+ */
+ public Object extractPayloadFromJMSMessage(Message msg) {
+ try {
+ if (msg.getBooleanProperty(JMSBindingConstants.FAULT_PROPERTY)) {
+ throw new ServiceRuntimeException("remote service exception, see nested exception",(Throwable)((ObjectMessage)msg).getObject());
+ }
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ return extractPayload(msg);
+ }
+
+ public Message createFaultMessage(Session session, Throwable o) {
+ try {
+
+ ObjectMessage message = session.createObjectMessage();
+ message.setObject(o);
+ message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true);
+ return message;
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+ protected abstract Object[] extractPayload(Message msg);
+
+ protected abstract Message createJMSMessage(Session session, Object o);
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingInvoker.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingInvoker.java
new file mode 100644
index 0000000000..e5d14dd42f
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingInvoker.java
@@ -0,0 +1,273 @@
+/*
+ * 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.jms.provider;
+
+import java.lang.reflect.InvocationTargetException;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.Invoker;
+import org.apache.tuscany.sca.invocation.DataExchangeSemantics;
+import org.osoa.sca.ServiceRuntimeException;
+
+/**
+ * Interceptor for the JMS binding.
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSBindingInvoker implements Invoker, DataExchangeSemantics {
+
+ protected Operation operation;
+ protected String operationName;
+
+ protected JMSBinding jmsBinding;
+ protected JMSResourceFactory jmsResourceFactory;
+ protected JMSMessageProcessor requestMessageProcessor;
+ protected JMSMessageProcessor responseMessageProcessor;
+ protected Destination requestDest;
+ protected Destination replyDest;
+
+ public JMSBindingInvoker(JMSBinding jmsBinding, Operation operation, JMSResourceFactory jmsResourceFactory) {
+
+ this.operation = operation;
+ operationName = operation.getName();
+
+ this.jmsBinding = jmsBinding;
+ this.jmsResourceFactory = jmsResourceFactory;
+ requestMessageProcessor = jmsBinding.getRequestMessageProcessor();
+ responseMessageProcessor = jmsBinding.getResponseMessageProcessor();
+ try {
+ requestDest = lookupDestination();
+ replyDest = lookupResponseDestination();
+ } catch (NamingException e) {
+ throw new JMSBindingException(e);
+ }
+
+ }
+
+ /**
+ * Looks up the Destination Queue for the JMS Binding
+ *
+ * @return The Destination Queue
+ * @throws NamingException Failed to lookup Destination Queue
+ * @throws JMSBindingException Failed to lookup Destination Queue
+ * @see #lookupDestinationQueue(boolean)
+ */
+ private Destination lookupDestination() throws NamingException, JMSBindingException {
+ return lookupDestinationQueue(false);
+ }
+
+ /**
+ * Looks up the Destination Response Queue for the JMS Binding
+ *
+ * @return The Destination Response Queue
+ * @throws NamingException Failed to lookup Destination Response Queue
+ * @throws JMSBindingException Failed to lookup Destination Response Queue
+ * @see #lookupDestinationQueue(boolean)
+ */
+ private Destination lookupResponseDestination() throws NamingException, JMSBindingException {
+ return lookupDestinationQueue(true);
+ }
+
+ /**
+ * Looks up the Destination Queue for the JMS Binding.
+ * <p>
+ * What happens in the look up will depend on the create mode specified for the JMS Binding:
+ * <ul>
+ * <li>always - the JMS queue is always created. It is an error if the queue already exists
+ * <li>ifnotexist - the JMS queue is created if it does not exist. It is not an error if the queue already exists
+ * <li>never - the JMS queue is never created. It is an error if the queue does not exist
+ * </ul>
+ * See the SCA JMS Binding specification for more information.
+ * <p>
+ *
+ * @param isReponseQueue <code>true</code> if we are creating a response queue. <code>false</code> if we are
+ * creating a request queue
+ * @return The Destination queue.
+ * @throws NamingException Failed to lookup JMS queue
+ * @throws JMSBindingException Failed to lookup JMS Queue. Probable cause is that the JMS queue's current
+ * existence/non-existence is not compatible with the create mode specified on the binding
+ */
+ private Destination lookupDestinationQueue(boolean isReponseQueue) throws NamingException, JMSBindingException {
+ String queueName;
+ String queueType;
+ String qCreateMode;
+ if (isReponseQueue) {
+ queueName = jmsBinding.getResponseDestinationName();
+ queueType = "JMS Response Destination ";
+ qCreateMode = jmsBinding.getResponseDestinationCreate();
+ if (JMSBindingConstants.DEFAULT_RESPONSE_DESTINATION_NAME.equals(queueName)) {
+ return null;
+ }
+ } else {
+ queueName = jmsBinding.getDestinationName();
+ queueType = "JMS Destination ";
+ qCreateMode = jmsBinding.getDestinationCreate();
+ }
+
+ Destination dest = jmsResourceFactory.lookupDestination(queueName);
+
+ if (qCreateMode.equals(JMSBindingConstants.CREATE_ALWAYS)) {
+ // In this mode, the queue must not already exist as we are creating it
+ if (dest != null) {
+ throw new JMSBindingException(queueType + queueName
+ + " already exists but has create mode of \""
+ + qCreateMode
+ + "\" while registering binding "
+ + jmsBinding.getName()
+ + " invoker");
+ }
+ // Create the queue
+ dest = jmsResourceFactory.createDestination(queueName);
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_IF_NOT_EXIST)) {
+ // In this mode, the queue may nor may not exist. It will be created if it does not exist
+ if (dest == null) {
+ dest = jmsResourceFactory.createDestination(queueName);
+ }
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_NEVER)) {
+ // In this mode, the queue must have already been created.
+ if (dest == null) {
+ throw new JMSBindingException(queueType + queueName
+ + " not found but create mode of \""
+ + qCreateMode
+ + "\" while registering binding "
+ + jmsBinding.getName()
+ + " invoker");
+ }
+ }
+
+ // Make sure we ended up with a queue
+ if (dest == null) {
+ throw new JMSBindingException(queueType + queueName
+ + " not found with create mode of \""
+ + qCreateMode
+ + "\" while registering binding "
+ + jmsBinding.getName()
+ + " invoker");
+ }
+
+ return dest;
+ }
+
+ public org.apache.tuscany.sca.invocation.Message invoke(org.apache.tuscany.sca.invocation.Message msg) {
+ try {
+ Object resp = invokeTarget((Object[])msg.getBody(), (short)0);
+ msg.setBody(resp);
+ } catch (InvocationTargetException e) {
+ msg.setFaultBody(e.getCause());
+ } catch (ServiceRuntimeException e) {
+ if (e.getCause() instanceof InvocationTargetException) {
+ if ((e.getCause().getCause() instanceof RuntimeException)) {
+ msg.setFaultBody(e.getCause());
+ } else {
+ msg.setFaultBody(e.getCause().getCause());
+ }
+ } else {
+ msg.setFaultBody(e);
+ }
+ } catch (Throwable e) {
+ msg.setFaultBody(e);
+ }
+ return msg;
+ }
+
+ public Object invokeTarget(Object payload, final short sequence) throws InvocationTargetException {
+ try {
+ Session session = jmsResourceFactory.createSession();
+ try {
+
+ Destination replyToDest;
+ if (operation.isNonBlocking()) {
+ replyToDest = null;
+ } else {
+ replyToDest = (replyDest != null) ? replyDest : session.createTemporaryQueue();
+ }
+
+ Message requestMsg = sendRequest((Object[])payload, session, replyToDest);
+ if (replyToDest == null) {
+ return null;
+ } else {
+ Message replyMsg = receiveReply(session, replyToDest, requestMsg.getJMSMessageID());
+ return ((Object[])responseMessageProcessor.extractPayloadFromJMSMessage(replyMsg))[0];
+ }
+
+ } finally {
+ session.close();
+ }
+ } catch (JMSException e) {
+ throw new InvocationTargetException(e);
+ } catch (NamingException e) {
+ throw new InvocationTargetException(e);
+ }
+ }
+
+ protected Message sendRequest(Object payload, Session session, Destination replyToDest) throws JMSException {
+
+ Message requestMsg = requestMessageProcessor.insertPayloadIntoJMSMessage(session, payload);
+
+ requestMsg.setJMSDeliveryMode(jmsBinding.getDeliveryMode());
+ requestMsg.setJMSPriority(jmsBinding.getPriority());
+
+ requestMessageProcessor.setOperationName(operationName, requestMsg);
+ requestMsg.setJMSReplyTo(replyToDest);
+
+ MessageProducer producer = session.createProducer(requestDest);
+ try {
+ producer.send(requestMsg);
+ } finally {
+ producer.close();
+ }
+ return requestMsg;
+ }
+
+ protected Message receiveReply(Session session, Destination replyToDest, String requestMsgId) throws JMSException,
+ NamingException {
+ String msgSelector = "JMSCorrelationID = '" + requestMsgId + "'";
+ MessageConsumer consumer = session.createConsumer(replyToDest, msgSelector);
+ Message replyMsg;
+ try {
+ jmsResourceFactory.startConnection();
+ replyMsg = consumer.receive(jmsBinding.getTimeToLive());
+ } finally {
+ consumer.close();
+ }
+ if (replyMsg == null) {
+ throw new JMSBindingException("No reply message received on " + replyToDest + " for message id " + requestMsgId);
+ }
+ return replyMsg;
+ }
+
+ public boolean allowsPassByReference() {
+ // JMS always pass by value
+ return true;
+ }
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingListener.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingListener.java
new file mode 100644
index 0000000000..61c4ec65fc
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingListener.java
@@ -0,0 +1,169 @@
+/*
+ * 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.jms.provider;
+
+import java.lang.reflect.InvocationTargetException;
+import java.util.List;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageListener;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+
+/**
+ * Listener for the JMSBinding.
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSBindingListener implements MessageListener {
+
+ private static final Logger logger = Logger.getLogger(JMSBindingListener.class.getName());
+
+ private static final String ON_MESSAGE_METHOD_NAME = "onMessage";
+ private JMSBinding jmsBinding;
+ private JMSResourceFactory jmsResourceFactory;
+ private RuntimeComponentService service;
+ private JMSMessageProcessor requestMessageProcessor;
+ private JMSMessageProcessor responseMessageProcessor;
+ private String correlationScheme;
+
+ public JMSBindingListener(JMSBinding jmsBinding,
+ JMSResourceFactory jmsResourceFactory,
+ RuntimeComponentService service) throws NamingException {
+ this.jmsBinding = jmsBinding;
+ this.jmsResourceFactory = jmsResourceFactory;
+ this.service = service;
+ requestMessageProcessor = jmsBinding.getRequestMessageProcessor();
+ responseMessageProcessor = jmsBinding.getResponseMessageProcessor();
+ correlationScheme = jmsBinding.getCorrelationScheme();
+ }
+
+ public void onMessage(Message requestJMSMsg) {
+ logger.log(Level.FINE, "JMS service '" + service.getName() + "' received message " + requestJMSMsg);
+ try {
+ Object responsePayload = invokeService(requestJMSMsg);
+ sendReply(requestJMSMsg, responsePayload, false);
+ } catch (Throwable e) {
+ logger.log(Level.SEVERE, "Exception invoking service '" + service.getName(), e);
+ sendReply(requestJMSMsg, e, true);
+ }
+ }
+
+ /**
+ * Turn the JMS message back into a Tuscany message and invoke the target component
+ *
+ * @param requestJMSMsg
+ * @return
+ * @throws JMSException
+ * @throws InvocationTargetException
+ */
+ protected Object invokeService(Message requestJMSMsg) throws JMSException, InvocationTargetException {
+
+ String operationName = requestMessageProcessor.getOperationName(requestJMSMsg);
+ Object requestPayload = requestMessageProcessor.extractPayloadFromJMSMessage(requestJMSMsg);
+
+ List<Operation> opList = service.getInterfaceContract().getInterface().getOperations();
+
+ Operation operation = null;
+
+ if (opList.size() == 1) {
+ // SCA JMS Binding Specification - Rule 1.5.1 line 203
+ operation = opList.get(0);
+ } else if (operationName != null) {
+ // SCA JMS Binding Specification - Rule 1.5.1 line 205
+ for (Operation op : opList) {
+ if (op.getName().equals(operationName)) {
+ operation = op;
+ break;
+ }
+ }
+ } else {
+ // SCA JMS Binding Specification - Rule 1.5.1 line 207
+ for (Operation op : opList) {
+ if (op.getName().equals(ON_MESSAGE_METHOD_NAME)) {
+ operation = op;
+ break;
+ }
+ }
+ }
+
+ if (operation != null) {
+ return service.getRuntimeWire(jmsBinding).invoke(operation, (Object[])requestPayload);
+ } else {
+ throw new JMSBindingException("Can't find operation " + (operationName != null ? operationName
+ : ON_MESSAGE_METHOD_NAME));
+ }
+
+ }
+
+ protected void sendReply(Message requestJMSMsg, Object responsePayload, boolean isFault) {
+ try {
+
+ if (requestJMSMsg.getJMSReplyTo() == null) {
+ // assume no reply is expected
+ if (responsePayload != null) {
+ logger.log(Level.FINE, "JMS service '" + service.getName() + "' dropped response as request has no replyTo");
+ }
+ return;
+ }
+
+ Session session = jmsResourceFactory.createSession();
+ Message replyJMSMsg;
+ if (isFault) {
+ replyJMSMsg = responseMessageProcessor.createFaultMessage(session, (Throwable)responsePayload);
+ } else {
+ replyJMSMsg = responseMessageProcessor.insertPayloadIntoJMSMessage(session, responsePayload);
+ }
+
+ replyJMSMsg.setJMSDeliveryMode(requestJMSMsg.getJMSDeliveryMode());
+ replyJMSMsg.setJMSPriority(requestJMSMsg.getJMSPriority());
+
+ if (correlationScheme == null || JMSBindingConstants.CORRELATE_MSG_ID.equalsIgnoreCase(correlationScheme)) {
+ replyJMSMsg.setJMSCorrelationID(requestJMSMsg.getJMSMessageID());
+ } else if (JMSBindingConstants.CORRELATE_CORRELATION_ID.equalsIgnoreCase(correlationScheme)) {
+ replyJMSMsg.setJMSCorrelationID(requestJMSMsg.getJMSCorrelationID());
+ }
+
+ Destination destination = requestJMSMsg.getJMSReplyTo();
+ MessageProducer producer = session.createProducer(destination);
+
+ producer.send(replyJMSMsg);
+
+ producer.close();
+ session.close();
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ } catch (NamingException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java
new file mode 100644
index 0000000000..57fc0e1230
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java
@@ -0,0 +1,62 @@
+/*
+ * 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.jms.provider;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.core.ExtensionPointRegistry;
+import org.apache.tuscany.sca.core.UtilityExtensionPoint;
+import org.apache.tuscany.sca.provider.BindingProviderFactory;
+import org.apache.tuscany.sca.provider.ReferenceBindingProvider;
+import org.apache.tuscany.sca.provider.ServiceBindingProvider;
+import org.apache.tuscany.sca.runtime.RuntimeComponent;
+import org.apache.tuscany.sca.runtime.RuntimeComponentReference;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+import org.apache.tuscany.sca.work.WorkScheduler;
+
+/**
+ * A factory from creating the JMS binding provider.
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSBindingProviderFactory implements BindingProviderFactory<JMSBinding> {
+
+ private WorkScheduler workScheduler;
+
+ public JMSBindingProviderFactory(ExtensionPointRegistry extensionPoints) {
+ UtilityExtensionPoint utilities = extensionPoints.getExtensionPoint(UtilityExtensionPoint.class);
+ workScheduler = utilities.getUtility(WorkScheduler.class);
+ }
+
+ public ReferenceBindingProvider createReferenceBindingProvider(RuntimeComponent component,
+ RuntimeComponentReference reference,
+ JMSBinding binding) {
+ return new JMSBindingReferenceBindingProvider(component, reference, binding);
+ }
+
+ public ServiceBindingProvider createServiceBindingProvider(RuntimeComponent component,
+ RuntimeComponentService service,
+ JMSBinding binding) {
+ return new JMSBindingServiceBindingProvider(component, service, binding, workScheduler);
+ }
+
+ public Class<JMSBinding> getModelType() {
+ return JMSBinding.class;
+ }
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java
new file mode 100644
index 0000000000..4920e92ae5
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java
@@ -0,0 +1,142 @@
+/*
+ * 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.jms.provider;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import javax.jms.JMSException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.interfacedef.Interface;
+import org.apache.tuscany.sca.interfacedef.InterfaceContract;
+import org.apache.tuscany.sca.interfacedef.Operation;
+import org.apache.tuscany.sca.invocation.Invoker;
+import org.apache.tuscany.sca.provider.ReferenceBindingProvider;
+import org.apache.tuscany.sca.runtime.RuntimeComponent;
+import org.apache.tuscany.sca.runtime.RuntimeComponentReference;
+
+/**
+ * Implementation of the JMS reference binding provider.
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSBindingReferenceBindingProvider implements ReferenceBindingProvider {
+
+ private RuntimeComponentReference reference;
+ private JMSBinding jmsBinding;
+ private List<JMSBindingInvoker> jmsBindingInvokers = new ArrayList<JMSBindingInvoker>();
+ private JMSResourceFactory jmsResourceFactory;
+
+ public JMSBindingReferenceBindingProvider(RuntimeComponent component,
+ RuntimeComponentReference reference,
+ JMSBinding binding) {
+ this.reference = reference;
+ this.jmsBinding = binding;
+ jmsResourceFactory = new JMSResourceFactory(binding.getConnectionFactoryName(), binding.getInitialContextFactoryName(), binding.getJndiURL());
+
+ if (XMLTextMessageProcessor.class.isAssignableFrom(jmsBinding.getRequestMessageProcessor().getClass())) {
+ setXMLDataBinding(reference);
+ }
+
+ }
+
+ protected void setXMLDataBinding(RuntimeComponentReference reference) {
+ try {
+ InterfaceContract ic = (InterfaceContract)reference.getInterfaceContract().clone();
+
+ Interface ii = (Interface)ic.getInterface().clone();
+ ii.resetDataBinding("org.apache.axiom.om.OMElement");
+ ic.setInterface(ii);
+ reference.setInterfaceContract(ic);
+
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public Invoker createInvoker(Operation operation) {
+
+ if (jmsBinding.getDestinationName().equals(JMSBindingConstants.DEFAULT_DESTINATION_NAME)) {
+ throw new JMSBindingException("No destination specified for reference " + reference.getName());
+ }
+
+ /* The following doesn't work as I can't get to the
+ * target list on the composite reference
+ // if the default destination queue name is set
+ // set the destination queue name to the wired service name
+ // so that any wires can be assured a unique endpoint.
+
+ if (jmsBinding.getDestinationName().equals(JMSBindingConstants.DEFAULT_DESTINATION_NAME)){
+ // get the name of the target service
+ List<ComponentService> targets = reference.getTargets();
+
+ if (targets.size() < 1){
+ throw new JMSBindingException("No target specified for reference " +
+ reference.getName() +
+ " so destination queue name can't be determined");
+ }
+
+ if (targets.size() > 1){
+ throw new JMSBindingException("More than one target specified for reference " +
+ reference.getName() +
+ " so destination queue name can't be determined");
+ }
+
+ ComponentService service = targets.get(0);
+ jmsBinding.setDestinationName(service.getName());
+ }
+
+
+ // if the default response queue name is set
+ // set the response queue to the names of this
+ // reference
+ if (jmsBinding.getResponseDestinationName().equals(JMSBindingConstants.DEFAULT_RESPONSE_DESTINATION_NAME)){
+ jmsBinding.setResponseDestinationName(reference.getName());
+ }
+ */
+ JMSBindingInvoker invoker = new JMSBindingInvoker(jmsBinding, operation, jmsResourceFactory);
+ jmsBindingInvokers.add(invoker);
+ return invoker;
+ }
+
+ public boolean supportsOneWayInvocation() {
+ return true;
+ }
+
+ public InterfaceContract getBindingInterfaceContract() {
+ return reference.getInterfaceContract();
+ }
+
+ public void start() {
+
+ }
+
+ public void stop() {
+ try {
+ jmsResourceFactory.closeConnection();
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java
new file mode 100644
index 0000000000..15a4d0ada2
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java
@@ -0,0 +1,227 @@
+/*
+ * 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.jms.provider;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.Session;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.interfacedef.Interface;
+import org.apache.tuscany.sca.interfacedef.InterfaceContract;
+import org.apache.tuscany.sca.provider.ServiceBindingProvider;
+import org.apache.tuscany.sca.runtime.RuntimeComponent;
+import org.apache.tuscany.sca.runtime.RuntimeComponentService;
+import org.apache.tuscany.sca.work.WorkScheduler;
+
+/**
+ * Implementation of the JMS service binding provider.
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSBindingServiceBindingProvider implements ServiceBindingProvider {
+ private static final Logger logger = Logger.getLogger(JMSBindingServiceBindingProvider.class.getName());
+
+ private RuntimeComponentService service;
+ private JMSBinding jmsBinding;
+ private JMSResourceFactory jmsResourceFactory;
+ private MessageConsumer consumer;
+ private WorkScheduler workScheduler;
+ private boolean running;
+
+ public JMSBindingServiceBindingProvider(RuntimeComponent component,
+ RuntimeComponentService service,
+ JMSBinding binding,
+ WorkScheduler workScheduler) {
+ this.service = service;
+ this.jmsBinding = binding;
+ this.workScheduler = workScheduler;
+
+ jmsResourceFactory = new JMSResourceFactory(binding.getConnectionFactoryName(), binding.getInitialContextFactoryName(), binding.getJndiURL());
+
+ if (jmsBinding.getDestinationName().equals(JMSBindingConstants.DEFAULT_DESTINATION_NAME)) {
+ // use the SCA service name as the default destination name
+ jmsBinding.setDestinationName(service.getName());
+ }
+
+ if (XMLTextMessageProcessor.class.isAssignableFrom(jmsBinding.getRequestMessageProcessor().getClass())) {
+ setXMLDataBinding(service);
+ }
+
+ }
+
+ protected void setXMLDataBinding(RuntimeComponentService service) {
+ if (service.getInterfaceContract() != null) {
+ try {
+ InterfaceContract ic = (InterfaceContract)service.getInterfaceContract().clone();
+
+ Interface ii = (Interface)ic.getInterface().clone();
+ ii.resetDataBinding("org.apache.axiom.om.OMElement");
+ ic.setInterface(ii);
+ service.setInterfaceContract(ic);
+
+ } catch (CloneNotSupportedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ }
+
+ public InterfaceContract getBindingInterfaceContract() {
+ return service.getInterfaceContract();
+ }
+
+ public boolean supportsOneWayInvocation() {
+ return true;
+ }
+
+ public void start() {
+ this.running = true;
+
+ try {
+ registerListerner();
+ } catch (Exception e) {
+ throw new JMSBindingException("Error starting JMSServiceBinding", e);
+ }
+ }
+
+ public void stop() {
+ this.running = false;
+ try {
+ consumer.close();
+ jmsResourceFactory.closeConnection();
+ } catch (Exception e) {
+ throw new JMSBindingException("Error stopping JMSServiceBinding", e);
+ }
+ }
+
+ private void registerListerner() throws NamingException, JMSException {
+
+ Session session = jmsResourceFactory.createSession();
+ Destination destination = lookupDestinationQueue();
+
+ consumer = session.createConsumer(destination);
+
+ final JMSBindingListener listener = new JMSBindingListener(jmsBinding, jmsResourceFactory, service);
+ try {
+
+ consumer.setMessageListener(listener);
+ jmsResourceFactory.startConnection();
+
+ } catch (javax.jms.IllegalStateException e) {
+
+ // setMessageListener not allowed in JEE container so use Tuscany threads
+
+ jmsResourceFactory.startConnection();
+ workScheduler.scheduleWork(new Runnable() {
+ public void run() {
+ try {
+ while (running) {
+ final Message msg = consumer.receive();
+ workScheduler.scheduleWork(new Runnable() {
+ public void run() {
+ try {
+ listener.onMessage(msg);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }});
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }});
+ }
+ logger.log(Level.INFO, "JMS service '" + service.getName() + "' listening on destination " + jmsBinding.getDestinationName());
+ }
+
+ /**
+ * Looks up the Destination Queue for the JMS Binding.
+ * <p>
+ * What happens in the look up will depend on the create mode specified for the JMS Binding:
+ * <ul>
+ * <li>always - the JMS queue is always created. It is an error if the queue already exists
+ * <li>ifnotexist - the JMS queue is created if it does not exist. It is not an error if the queue already exists
+ * <li>never - the JMS queue is never created. It is an error if the queue does not exist
+ * </ul>
+ * See the SCA JMS Binding specification for more information.
+ * <p>
+ *
+ * @return The Destination queue.
+ * @throws NamingException Failed to lookup JMS queue
+ * @throws JMSBindingException Failed to lookup JMS Queue. Probable cause is that the JMS queue's current
+ * existence/non-existence is not compatible with the create mode specified on the binding
+ */
+ private Destination lookupDestinationQueue() throws NamingException, JMSBindingException {
+ Destination destination = jmsResourceFactory.lookupDestination(jmsBinding.getDestinationName());
+
+ String qCreateMode = jmsBinding.getDestinationCreate();
+ if (qCreateMode.equals(JMSBindingConstants.CREATE_ALWAYS)) {
+ // In this mode, the queue must not already exist as we are creating it
+ if (destination != null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " already exists but has create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + service.getName()
+ + " listener");
+ }
+
+ // Create the queue
+ destination = jmsResourceFactory.createDestination(jmsBinding.getDestinationName());
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_IF_NOT_EXIST)) {
+ // In this mode, the queue may nor may not exist. It will be created if it does not exist
+ if (destination == null) {
+ destination = jmsResourceFactory.createDestination(jmsBinding.getDestinationName());
+ }
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_NEVER)) {
+ // In this mode, the queue must have already been created.
+ if (destination == null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " not found but create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + service.getName()
+ + " listener");
+ }
+ }
+
+ // Make sure we ended up with a queue
+ if (destination == null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " not found with create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + service.getName()
+ + " listener");
+ }
+
+ return destination;
+ }
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java
new file mode 100644
index 0000000000..733cebacac
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java
@@ -0,0 +1,55 @@
+/*
+ * 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.jms.provider;
+
+import javax.jms.Message;
+import javax.jms.Session;
+
+/**
+ * Interface for a component that does operation selection and message payload processing
+ *
+ * @version $Rev$ $Date$
+ */
+public interface JMSMessageProcessor {
+
+ /**
+ * Get the operation name from a JMS Message
+ */
+ String getOperationName(Message message);
+
+ /**
+ * Set the operation name on a JMS Message
+ */
+ void setOperationName(String operationName, Message message);
+
+ /**
+ * Extracts the payload from a JMS Message
+ */
+ Object extractPayloadFromJMSMessage(Message msg);
+
+ /**
+ * Create a JMS Message containing the payload
+ */
+ Message insertPayloadIntoJMSMessage(Session session, Object payload);
+
+ /**
+ * Create a JMS Message for reporting an exception
+ */
+ Message createFaultMessage(Session session, Throwable responsePayload);
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java
new file mode 100644
index 0000000000..23a77d4a73
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java
@@ -0,0 +1,176 @@
+/*
+ * 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.jms.provider;
+
+import java.util.Properties;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+
+/**
+ * Abstracts away any JMS provide specific feature from the JMS binding
+ *
+ * @version $Rev$ $Date$
+ */
+public class JMSResourceFactory {
+
+ private String initialContextFactoryName;
+ private String connectionFactoryName = "ConnectionFactory";
+ private String jndiURL;
+
+ private Connection connection;
+ private Context context;
+ private boolean isConnectionStarted;
+
+ public JMSResourceFactory(String connectionFactoryName, String initialContextFactoryName, String jndiURL) {
+ if (connectionFactoryName != null && connectionFactoryName.trim().length() > 0) {
+ this.connectionFactoryName = connectionFactoryName.trim();
+ }
+ if (initialContextFactoryName != null && initialContextFactoryName.trim().length() > 0) {
+ this.initialContextFactoryName = initialContextFactoryName.trim();
+ }
+ if (jndiURL != null) {
+ this.jndiURL = jndiURL.trim();
+ }
+ }
+
+ /*
+ * This is a simple implementation where a connection is created per binding Ideally the resource factory should be
+ * able to leverage the host environment to provide connection pooling if it can. E.g. if Tuscany is running inside
+ * an AppServer Then we could leverage the JMS resources it provides
+ *
+ * @see org.apache.tuscany.binding.jms.JMSResourceFactory#getConnection()
+ */
+ public Connection getConnection() throws NamingException, JMSException {
+ if (connection == null) {
+ createConnection();
+ }
+ return connection;
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.JMSResourceFactory#createSession()
+ */
+ public Session createSession() throws JMSException, NamingException {
+ return getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.JMSResourceFactory#startConnection()
+ */
+ public void startConnection() throws JMSException, NamingException {
+ if (!isConnectionStarted) {
+ getConnection().start();
+ isConnectionStarted = true;
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.tuscany.binding.jms.JMSResourceFactory#closeConnection()
+ */
+ public void closeConnection() throws JMSException {
+ if (connection != null) {
+ connection.close();
+ }
+ }
+
+ private void createConnection() throws NamingException, JMSException {
+ ConnectionFactory connectionFactory = (ConnectionFactory)jndiLookUp(connectionFactoryName);
+ if (connectionFactory == null) {
+ throw new JMSBindingException("connection factory not found: " + connectionFactoryName);
+ }
+ connection = connectionFactory.createConnection();
+ }
+
+ private synchronized Context getInitialContext() throws NamingException {
+ if (context == null) {
+ Properties props = new Properties();
+
+ if (initialContextFactoryName != null) {
+ props.setProperty(Context.INITIAL_CONTEXT_FACTORY, initialContextFactoryName);
+ }
+ if (jndiURL != null) {
+ props.setProperty(Context.PROVIDER_URL, jndiURL);
+ }
+
+ initJREEnvironment(props);
+
+ context = new InitialContext(props);
+ }
+ return context;
+ }
+
+ /**
+ * If using the WAS JMS Client with a non-IBM JRE then an additional
+ * environment property needs to be set to initialize the ORB correctly.
+ * See: http://www-1.ibm.com/support/docview.wss?uid=swg24012804
+ */
+ private void initJREEnvironment(Properties props) {
+ if ("com.ibm.websphere.naming.WsnInitialContextFactory".equals(props.get(Context.INITIAL_CONTEXT_FACTORY))) {
+ String vendor = System.getProperty("java.vendor");
+ if (vendor == null || !vendor.contains("IBM")) {
+ props.setProperty("com.ibm.CORBA.ORBInit","com.ibm.ws.sib.client.ORB");
+ }
+ }
+ }
+
+ public Destination lookupDestination(String jndiName) throws NamingException {
+ return (Destination)jndiLookUp(jndiName);
+ }
+
+ /**
+ * You can create a destination in ActiveMQ (and have it appear in JNDI) by putting "dynamicQueues/" in front of the
+ * queue name being looked up
+ */
+ public Destination createDestination(String jndiName) throws NamingException {
+ return lookupDestination("dynamicQueues/" + jndiName);
+ }
+
+ protected Object jndiLookUp(String name) {
+ Object o = null;
+ try {
+ o = getInitialContext().lookup("java:comp/env/" + name);
+ } catch (NamingException ex) {
+ // ignore
+ }
+ if (o == null) {
+ try {
+ o = getInitialContext().lookup(name);
+ } catch (NamingException ex) {
+ ex.printStackTrace();
+ // ignore
+ }
+ }
+ return o;
+ }
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java
new file mode 100644
index 0000000000..e63c1ec19b
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java
@@ -0,0 +1,70 @@
+/*
+ * 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.jms.provider;
+
+import java.io.Serializable;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.ObjectMessage;
+import javax.jms.Session;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+
+/**
+ * MessageProcessor for sending/receiving Serializable objects with
+ * the JMSBinding.
+ *
+ * @version $Rev$ $Date$
+ */
+public class ObjectMessageProcessor extends AbstractMessageProcessor {
+
+ public ObjectMessageProcessor(JMSBinding jmsBinding) {
+ super(jmsBinding);
+ }
+
+ @Override
+ protected Object[] extractPayload(Message msg) {
+ try {
+
+ return new Object[] {((ObjectMessage)msg).getObject()};
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+ protected Message createJMSMessage(Session session, Object o) {
+ try {
+
+ if (!(o instanceof Serializable)) {
+ throw new IllegalStateException("JMS ObjectMessage payload not Serializable: " + o);
+ }
+
+ ObjectMessage message = session.createObjectMessage();
+ message.setObject((Serializable)o);
+ return message;
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java
new file mode 100644
index 0000000000..cf9f6ce88b
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java
@@ -0,0 +1,69 @@
+/*
+ * 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.jms.provider;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+
+/**
+ * MessageProcessor for sending/receiving javax.jms.TextMessage with
+ * the JMSBinding.
+ *
+ * @version $Rev$ $Date$
+ */
+public class TextMessageProcessor extends AbstractMessageProcessor {
+
+ public TextMessageProcessor(JMSBinding jmsBinding) {
+ super(jmsBinding);
+ }
+
+ @Override
+ protected Object[] extractPayload(Message msg) {
+ try {
+
+ if (!(msg instanceof TextMessage)) {
+ throw new IllegalStateException("expecting JMS TextMessage: " + msg);
+ }
+
+ return new Object[]{((TextMessage)msg).getText()};
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+ @Override
+ protected Message createJMSMessage(Session session, Object o) {
+ try {
+
+ TextMessage message = session.createTextMessage();
+ message.setText(String.valueOf(o));
+ return message;
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}
diff --git a/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java
new file mode 100644
index 0000000000..8f4237b4bf
--- /dev/null
+++ b/sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java
@@ -0,0 +1,91 @@
+/*
+ * 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.jms.provider;
+
+import java.io.StringReader;
+
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.Session;
+import javax.jms.TextMessage;
+import javax.xml.stream.XMLInputFactory;
+import javax.xml.stream.XMLStreamException;
+import javax.xml.stream.XMLStreamReader;
+
+import org.apache.axiom.om.OMElement;
+import org.apache.axiom.om.impl.builder.StAXOMBuilder;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+
+/**
+ * MessageProcessor for sending/receiving XML javax.jms.TextMessage with
+ * the JMSBinding.
+ *
+ * @version $Rev$ $Date$
+ */
+public class XMLTextMessageProcessor extends AbstractMessageProcessor {
+
+ public XMLTextMessageProcessor(JMSBinding jmsBinding) {
+ super(jmsBinding);
+ }
+
+ @Override
+ protected Object[] extractPayload(Message msg) {
+ try {
+
+ String xml = ((TextMessage)msg).getText();
+ Object[] os;
+ if (xml != null) {
+ XMLStreamReader reader = XMLInputFactory.newInstance().createXMLStreamReader(new StringReader(xml));
+ StAXOMBuilder builder = new StAXOMBuilder(reader);
+ os = new Object[] { builder.getDocumentElement() };
+ } else {
+ os = new Object[]{};
+ }
+ return os;
+
+ } catch (XMLStreamException e) {
+ throw new JMSBindingException(e);
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+ @Override
+ protected Message createJMSMessage(Session session, Object o) {
+ try {
+
+ TextMessage message = session.createTextMessage();
+
+ if (o instanceof OMElement) {
+ message.setText(o.toString());
+ } else if ((o instanceof Object[]) && ((Object[])o)[0] instanceof OMElement) {
+ message.setText(((Object[])o)[0].toString());
+ } else if (o != null) {
+ throw new IllegalStateException("expecting OMElement payload: " + o);
+ }
+
+ return message;
+
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}