diff options
Diffstat (limited to 'sandbox/sebastien/java/sca-node/modules/binding-jms/src/main/java/org/apache/tuscany/sca/binding/jms/provider')
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); + } + } + +} |