diff options
Diffstat (limited to 'sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider')
18 files changed, 2806 insertions, 0 deletions
diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java new file mode 100644 index 0000000000..6c6339c88e --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/AbstractMessageProcessor.java @@ -0,0 +1,161 @@ +/* + * 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.PrintWriter; +import java.io.StringWriter; +import java.lang.reflect.InvocationTargetException; +import java.util.logging.Logger; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.ObjectMessage; +import javax.jms.Session; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.oasisopen.sca.ServiceRuntimeException; + +/** + * Base MessageProcessor for the JMSBinding. + * + * @version $Rev$ $Date$ + */ +public abstract class AbstractMessageProcessor implements JMSMessageProcessor { + private static final Logger logger = Logger.getLogger(AbstractMessageProcessor.class.getName()); + + 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)) { + Object exc = ((ObjectMessage)msg).getObject(); + if (exc instanceof RuntimeException) { + throw new ServiceRuntimeException("remote service exception, see nested exception", (Throwable)exc); + } else { + return new InvocationTargetException((Throwable) exc); + } + } + } catch (JMSException e) { + throw new JMSBindingException(e); + } + return extractPayload(msg); + } + + public Message createFaultMessage(Session session, Throwable o) { + if (session == null) { + logger.fine("no response session to create fault message: " + String.valueOf(o)); + return null; + } + try { + + ObjectMessage message = session.createObjectMessage(); + if (o instanceof RuntimeException || o instanceof Error) { + int recursionKlugeDetector = 20; + Throwable rootCause = o; + Throwable deepRootCause = rootCause.getCause(); + do { + if (rootCause == deepRootCause) { + break; + } else if (deepRootCause != null) { + rootCause = deepRootCause; + } + + if (recursionKlugeDetector-- <= 0) { + break; + } + } while (deepRootCause != null); + + final StringWriter sw = new StringWriter(); + final PrintWriter pw = new PrintWriter(sw); + pw.print("Message = " + o.getMessage()); + StackTraceElement[] stackElements = o.getStackTrace(); + for (int i = 0; i < stackElements.length; i++) { + pw.print("\t>> \t at "); + pw.println(stackElements[i].toString()); + } + pw.flush(); + + message.setObject(new RuntimeException( sw.toString() )); + } else { + 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/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/BytesMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/BytesMessageProcessor.java new file mode 100644 index 0000000000..7b14d29dfa --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/BytesMessageProcessor.java @@ -0,0 +1,120 @@ +/* + * 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.Logger; + +import javax.jms.BytesMessage; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; + +/** + * MessageProcessor for sending/receiving javax.jms.BytesMessage with the JMSBinding. + * + * @version $Rev$ $Date$ + */ +public class BytesMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(AbstractMessageProcessor.class.getName()); + + public BytesMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + } + +/* TUSCANY-2967 - disable this change while we decide what to do and + * return faults as JMSObject messages to be consistent + * again with other wire formats + @Override + public Object extractPayloadFromJMSMessage(Message msg) { + byte [] bytes = (byte [])extractPayload(msg); + + try { + if (msg.getBooleanProperty(JMSBindingConstants.FAULT_PROPERTY)) { + return new InvocationTargetException(new ServiceRuntimeException(new String(bytes))); + } else { + return bytes; + } + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } +*/ + + @Override + protected Object extractPayload(Message msg) { + try { + + if (!(msg instanceof BytesMessage)) { + throw new IllegalStateException("expecting JMS BytesMessage: " + msg); + } + + long noOfBytes = ((BytesMessage)msg).getBodyLength(); + byte [] bytes = new byte[(int)noOfBytes]; + ((BytesMessage)msg).readBytes(bytes); + ((BytesMessage)msg).reset(); + return bytes; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + +/* TUSCANY-2967 - disable this change while we decide what to do and + * return faults as JMSObject messages to be consistent + * again with other wire formats + @Override + public Message createFaultMessage(Session session, Throwable o) { + try { + Message message = createJMSMessage(session, o.toString().getBytes()); + message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true); + return message; + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } +*/ + + @Override + protected Message createJMSMessage(Session session, Object o) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + + // TODO - an experiment. How to enforce a single + // byte array parameter + BytesMessage message = session.createBytesMessage(); + + if (o != null){ + message.writeBytes((byte[])o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultJMSResourceFactoryExtensionPoint.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultJMSResourceFactoryExtensionPoint.java new file mode 100644 index 0000000000..32531fa6c3 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultJMSResourceFactoryExtensionPoint.java @@ -0,0 +1,30 @@ +/* + * 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.JMSBinding; + +public class DefaultJMSResourceFactoryExtensionPoint implements JMSResourceFactoryExtensionPoint { + + public JMSResourceFactory createJMSResourceFactory(JMSBinding binding) { + return new JMSResourceFactoryImpl(binding.getConnectionFactoryName(), binding.getResponseConnectionFactoryName(), binding.getInitialContextFactoryName(), binding.getJndiURL()); + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultMessageProcessor.java new file mode 100644 index 0000000000..c538dd01da --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/DefaultMessageProcessor.java @@ -0,0 +1,302 @@ +/* + * 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.IOException; +import java.util.logging.Logger; + +import javax.jms.BytesMessage; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; +import javax.jms.TextMessage; +import javax.xml.namespace.QName; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.common.xml.dom.DOMHelper; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.interfacedef.util.FaultException; +import org.w3c.dom.Document; +import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.xml.sax.SAXException; + +/** + * MessageProcessor for sending/receiving XML over javax.jms.TextMessage or javax.jms.BytesMessage + * with the JMSBinding. + * This is very specific to the default wire format and is not tied into the usual hierarchy + * of message processors + * + * @version $Rev$ $Date$ + */ +public class DefaultMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(DefaultMessageProcessor.class.getName()); + + private DOMHelper domHelper; + + public DefaultMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + this.domHelper = DOMHelper.getInstance(registry); + } + + // inherited methods that don't do anything useful + @Override + protected Message createJMSMessage(Session session, Object o) { + // should not be used + return null; + } + + @Override + protected Object extractPayload(Message msg) { + // if it's not a text/bytes message or a fault then we don;t know what to do with it + return null; + } + + // TODO - This makes the assumption that whatever the text/bytes configuration of the + // jms binding, unchecked faults will be sent as bytes. + @Override + public Message createFaultMessage(Session session, Throwable o) { + return createFaultJMSBytesMessage(session, o); + } + + // handle text messages + + public Object extractPayloadFromJMSTextMessage(Message msg, Node wrapper) { + if (msg instanceof TextMessage) { + try { + String xml = ((TextMessage) msg).getText(); + + Object os; + if (xml != null && xml.length() > 0) { + os = domHelper.load(xml); + } else { + os = null; + } + + if (wrapper != null){ + //don't modify the original wrapper since it will be reused + //clone the wrapper + Node node = ((Node)os); + if (node == null) { + node = domHelper.newDocument(); + } + Element newWrapper = DOMHelper.createElement((Document)node, new QName(wrapper.getNamespaceURI(), wrapper.getLocalName())); + if (os != null){ + Node child = node.getFirstChild(); + newWrapper.appendChild(child); + } + return newWrapper; + } + + return os; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } catch (IOException e) { + throw new JMSBindingException(e); + } catch (SAXException e) { + throw new JMSBindingException(e); + } + } else { + // handle the non-text fault case + return super.extractPayloadFromJMSMessage(msg); + } + } + + public Message insertPayloadIntoJMSTextMessage(Session session, Object o, boolean unwrap) { + + try { + + TextMessage message = session.createTextMessage(); + + if (o instanceof Node) { + + if (unwrap){ + Node firstElement = ((Node)o).getFirstChild(); + if (firstElement == null ) { + message.setText(""); + } else { + message.setText(domHelper.saveAsString(firstElement)); + } + }else { + message.setText(domHelper.saveAsString((Node)o)); + } + } else if ((o instanceof Object[]) && ((Object[]) o)[0] instanceof Node) { + if (unwrap){ + Node firstElement = ((Node)((Object[]) o)[0]).getFirstChild(); + if (firstElement == null ) { + message.setText(null); + } else { + message.setText(domHelper.saveAsString(firstElement)); + } + }else { + message.setText(domHelper.saveAsString((Node)((Object[])o)[0])); + } + } else if (o != null) { + throw new IllegalStateException("expecting Node payload: " + o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + public Message createFaultJMSTextMessage(Session session, Throwable o) { + + if (session == null) { + logger.fine("no response session to create fault message: " + String.valueOf(o)); + return null; + } + if (o instanceof FaultException) { + try { + + TextMessage message = session.createTextMessage(); + message.setText(domHelper.saveAsString((Node)((FaultException)o).getFaultInfo())); + message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true); + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } else { + // handle the non XML fault case + return super.createFaultMessage(session, o); + } + } + + // handle bytes messages + + public Object extractPayloadFromJMSBytesMessage(Message msg, Node wrapper) { + + if (msg instanceof BytesMessage) { + try { + Object os; + + long noOfBytes = ((BytesMessage) msg).getBodyLength(); + byte[] bytes = new byte[(int) noOfBytes]; + ((BytesMessage) msg).readBytes(bytes); + ((BytesMessage)msg).reset(); + + if ((bytes != null) && (bytes.length > 0)) { + os = domHelper.load(new String(bytes)); + } else { + os = null; + } + + if (wrapper != null){ + //don't modify the original wrapper since it will be reused + //clone the wrapper + Node node = ((Node)os); + if (node == null) { + node = domHelper.newDocument(); + } + Element newWrapper = DOMHelper.createElement((Document)node, new QName(wrapper.getNamespaceURI(), wrapper.getLocalName())); + if (os != null){ + Node child = node.getFirstChild(); + newWrapper.appendChild(child); + } + return newWrapper; + } + + return os; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } catch (IOException e) { + throw new JMSBindingException(e); + } catch (SAXException e) { + throw new JMSBindingException(e); + } + } else { + // trap the non-bytes fault case + return super.extractPayloadFromJMSMessage(msg); + } + } + + public Message insertPayloadIntoJMSBytesMessage(Session session, Object o, boolean unwrap) { + + try { + + BytesMessage message = session.createBytesMessage(); + + + if (o instanceof Node) { + if (unwrap) { + Node firstElement = ((Node)o).getFirstChild(); + if (firstElement == null ) { + //do nothing, the message will just be set with a byte[0] + } else { + message.writeBytes(domHelper.saveAsString(firstElement).getBytes()); + } + + } else { + message.writeBytes(domHelper.saveAsString((Node)o).getBytes()); + } + + } else if ((o instanceof Object[]) && ((Object[]) o)[0] instanceof Node) { + if (unwrap){ + Node firstElement = ((Node)((Object[]) o)[0]).getFirstChild(); + if (firstElement == null ) { + //do nothing, the message will just be set with a byte[0] + } else { + message.writeBytes(domHelper.saveAsString(firstElement).getBytes()); + } + + }else { + message.writeBytes(domHelper.saveAsString((Node)((Object[]) o)[0]).getBytes()); + } + } else if (o != null) { + throw new IllegalStateException("expecting Node payload: " + o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + public Message createFaultJMSBytesMessage(Session session, Throwable o) { + + if (session == null) { + logger.fine("no response session to create fault message: " + String.valueOf(o)); + return null; + } + + if (o instanceof FaultException) { + try { + + BytesMessage message = session.createBytesMessage(); + String s = domHelper.saveAsString((Node)((FaultException)o).getFaultInfo()); + message.writeBytes(s.getBytes()); + message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true); + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } else { + return super.createFaultMessage(session, o); + } + } +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingAsyncResponseInvoker.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingAsyncResponseInvoker.java new file mode 100644 index 0000000000..f07e9de29f --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingAsyncResponseInvoker.java @@ -0,0 +1,43 @@ +/* + * 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.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.invocation.InvokerAsyncResponse; +import org.apache.tuscany.sca.invocation.Message; +import org.apache.tuscany.sca.runtime.RuntimeEndpoint; +import org.apache.tuscany.sca.runtime.RuntimeEndpointReference; + +/** + * @version $Rev$ $Date$ + */ +public class JMSBindingAsyncResponseInvoker implements InvokerAsyncResponse { + + RuntimeEndpoint endpoint; + + public JMSBindingAsyncResponseInvoker(ExtensionPointRegistry extensionPoints, + RuntimeEndpoint endpoint) { + this.endpoint = endpoint; + } // end constructor + + public void invokeAsyncResponse(Message msg) { + // TODO + } // end method invokeAsyncResponse +} // end class JMSBindingAsyncResponseInvoker
\ No newline at end of file diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java new file mode 100644 index 0000000000..15bd8713f2 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingProviderFactory.java @@ -0,0 +1,76 @@ +/* + * 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.JMSBinding; +import org.apache.tuscany.sca.binding.jms.host.DefaultJMSHostExtensionPoint; +import org.apache.tuscany.sca.binding.jms.host.JMSHostExtensionPoint; +import org.apache.tuscany.sca.binding.jms.host.JMSServiceListenerFactory; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +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.RuntimeEndpoint; +import org.apache.tuscany.sca.runtime.RuntimeEndpointReference; + +/** + * A factory from creating the JMS binding provider. + * + * @version $Rev$ $Date$ + */ +public class JMSBindingProviderFactory implements BindingProviderFactory<JMSBinding> { + + private ExtensionPointRegistry extensionPoints; + private JMSResourceFactoryExtensionPoint jmsRFEP; + private JMSServiceListenerFactory serviceListenerFactory; + + public JMSBindingProviderFactory(ExtensionPointRegistry extensionPoints) { + this.extensionPoints = extensionPoints; + + jmsRFEP = (JMSResourceFactoryExtensionPoint)extensionPoints.getExtensionPoint(JMSResourceFactoryExtensionPoint.class); + if (jmsRFEP == null) { + jmsRFEP = new DefaultJMSResourceFactoryExtensionPoint(); + extensionPoints.addExtensionPoint(jmsRFEP); + } + + JMSHostExtensionPoint jmsHostExtensionPoint = (JMSHostExtensionPoint)extensionPoints.getExtensionPoint(JMSHostExtensionPoint.class); + if (jmsHostExtensionPoint == null) { + jmsHostExtensionPoint = new DefaultJMSHostExtensionPoint(extensionPoints); + extensionPoints.addExtensionPoint(jmsHostExtensionPoint); + } + serviceListenerFactory = jmsHostExtensionPoint.getJMSServiceListenerFactory(); + } + + public ReferenceBindingProvider createReferenceBindingProvider(RuntimeEndpointReference endpointReference) { + JMSResourceFactory jmsRF = jmsRFEP.createJMSResourceFactory((JMSBinding)endpointReference.getBinding()); + return new JMSBindingReferenceBindingProvider(endpointReference, extensionPoints, jmsRF); + } + + public ServiceBindingProvider createServiceBindingProvider(RuntimeEndpoint endpoint) { + JMSBinding binding = (JMSBinding)endpoint.getBinding(); + JMSResourceFactory jmsRF = jmsRFEP.createJMSResourceFactory(binding); + return new JMSBindingServiceBindingProvider(extensionPoints, endpoint, serviceListenerFactory, extensionPoints, jmsRF); + } + + public Class<JMSBinding> getModelType() { + return JMSBinding.class; + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java new file mode 100644 index 0000000000..3471e68d6f --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingReferenceBindingProvider.java @@ -0,0 +1,219 @@ +/* + * 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.MessageListener; +import javax.naming.NamingException; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.binding.jms.headers.HeaderReferenceInterceptor; +import org.apache.tuscany.sca.binding.jms.host.AsyncResponseJMSServiceListener; +import org.apache.tuscany.sca.binding.jms.host.JMSAsyncResponseInvoker; +import org.apache.tuscany.sca.binding.jms.transport.TransportReferenceInterceptor; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.core.FactoryExtensionPoint; +import org.apache.tuscany.sca.core.UtilityExtensionPoint; +import org.apache.tuscany.sca.interfacedef.InterfaceContract; +import org.apache.tuscany.sca.interfacedef.Operation; +import org.apache.tuscany.sca.invocation.InvocationChain; +import org.apache.tuscany.sca.invocation.Invoker; +import org.apache.tuscany.sca.invocation.MessageFactory; +import org.apache.tuscany.sca.invocation.Phase; +import org.apache.tuscany.sca.provider.EndpointReferenceAsyncProvider; +import org.apache.tuscany.sca.provider.ProviderFactoryExtensionPoint; +import org.apache.tuscany.sca.provider.WireFormatProvider; +import org.apache.tuscany.sca.provider.WireFormatProviderFactory; +import org.apache.tuscany.sca.runtime.RuntimeComponentReference; +import org.apache.tuscany.sca.runtime.RuntimeEndpointReference; +import org.apache.tuscany.sca.work.WorkScheduler; + +/** + * Implementation of the JMS reference binding provider. + * + * This version supports native async service invocations + * + * @version $Rev$ $Date$ + */ +public class JMSBindingReferenceBindingProvider implements EndpointReferenceAsyncProvider { + + private RuntimeEndpointReference endpointReference; + private RuntimeComponentReference reference; + private JMSBinding jmsBinding; + private JMSResourceFactory jmsResourceFactory; + private InterfaceContract interfaceContract; + private ExtensionPointRegistry extensions; + + private ProviderFactoryExtensionPoint providerFactories; + + private WireFormatProviderFactory requestWireFormatProviderFactory; + private WireFormatProvider requestWireFormatProvider; + + private WireFormatProviderFactory responseWireFormatProviderFactory; + private WireFormatProvider responseWireFormatProvider; + + private AsyncResponseJMSServiceListener responseQueue = null; + + public JMSBindingReferenceBindingProvider(RuntimeEndpointReference endpointReference, ExtensionPointRegistry extensions, JMSResourceFactory jmsResourceFactory) { + this.endpointReference = endpointReference; + this.reference = (RuntimeComponentReference) endpointReference.getReference(); + this.jmsBinding = (JMSBinding) endpointReference.getBinding(); + this.extensions = extensions; + this.jmsResourceFactory = jmsResourceFactory; + + // Get the factories/providers for operation selection + this.providerFactories = extensions.getExtensionPoint(ProviderFactoryExtensionPoint.class); + + // Get the factories/providers for wire format + this.requestWireFormatProviderFactory = + (WireFormatProviderFactory)providerFactories.getProviderFactory(jmsBinding.getRequestWireFormat().getClass()); + if (this.requestWireFormatProviderFactory != null){ + this.requestWireFormatProvider = requestWireFormatProviderFactory.createReferenceWireFormatProvider(endpointReference); + } + + this.responseWireFormatProviderFactory = + (WireFormatProviderFactory)providerFactories.getProviderFactory(jmsBinding.getResponseWireFormat().getClass()); + if (this.responseWireFormatProviderFactory != null){ + this.responseWireFormatProvider = responseWireFormatProviderFactory.createReferenceWireFormatProvider(endpointReference); + } + + // create an interface contract that reflects both request and response + // wire formats + try { + interfaceContract = (InterfaceContract)reference.getInterfaceContract().clone(); + + requestWireFormatProvider.configureWireFormatInterfaceContract(interfaceContract); + responseWireFormatProvider.configureWireFormatInterfaceContract(interfaceContract); + } catch (CloneNotSupportedException ex){ + interfaceContract = reference.getInterfaceContract(); + } // end try + + // If the service is asyncInvocation, then create a fixed response location + if( endpointReference.isAsyncInvocation() ) { + String asyncCallbackName = endpointReference.getReference().getName() + "_asyncResponse"; + jmsBinding.setResponseDestinationName(asyncCallbackName); + } // end if + + } // end constructor + + public Invoker createInvoker(Operation operation) { + + if (jmsBinding.getDestinationName() == null) { + throw new JMSBindingException("No destination specified for reference " + reference.getName()); + } // end if + + if ( jmsBinding.getActivationSpecName() != null ) { + throw new JMSBindingException("Activation spec can not be specified on an SCA reference binding."); + } + Invoker invoker = null; + invoker = new RRBJMSBindingInvoker(operation, jmsResourceFactory, endpointReference); + + return invoker; + } // end method createInvoker + + public boolean supportsOneWayInvocation() { + return true; + } + + public InterfaceContract getBindingInterfaceContract() { + return interfaceContract; + } + + public void start() { + // If the reference is async invocation, then a response queue handler and associated JMS listener must be created + // and started + if (endpointReference.isAsyncInvocation()) { + // Create the JMS listener + FactoryExtensionPoint modelFactories = extensions.getExtensionPoint(FactoryExtensionPoint.class); + MessageFactory messageFactory = modelFactories.getFactory(MessageFactory.class); + MessageListener listener; + try { + listener = new JMSAsyncResponseInvoker(endpointReference, messageFactory, jmsResourceFactory); + } catch (NamingException e) { + throw new JMSBindingException("Unable to create JMSResponseInvoker", e); + } // end try + + // Create the response queue handler + UtilityExtensionPoint utilities = extensions.getExtensionPoint(UtilityExtensionPoint.class); + WorkScheduler workScheduler = utilities.getUtility(WorkScheduler.class); + + responseQueue = new AsyncResponseJMSServiceListener(listener, + jmsBinding.getResponseDestinationName(), + jmsBinding, workScheduler, jmsResourceFactory); + responseQueue.start(); + } // end if + + } // end method start + + public void stop() { + try { + if( responseQueue != null ) { + responseQueue.stop(); + } // end if + + jmsResourceFactory.closeConnection(); + jmsResourceFactory.closeResponseConnection(); + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } // end method stop + + /* + * set up the reference binding wire with the right set of jms reference + * interceptors + */ + public void configure() { + + InvocationChain bindingChain = endpointReference.getBindingInvocationChain(); + + // add transport interceptor + bindingChain.addInterceptor(Phase.REFERENCE_BINDING_TRANSPORT, + new TransportReferenceInterceptor(jmsBinding, + jmsResourceFactory, + endpointReference) ); + + // add request wire format + bindingChain.addInterceptor(requestWireFormatProvider.getPhase(), + requestWireFormatProvider.createInterceptor()); + + // add response wire format, but only add it if it's different from the request + if (!jmsBinding.getRequestWireFormat().equals(jmsBinding.getResponseWireFormat())){ + bindingChain.addInterceptor(responseWireFormatProvider.getPhase(), + responseWireFormatProvider.createInterceptor()); + } + + // add the header processor that comes after the wire formatter but before the + // policy interceptors + bindingChain.addInterceptor(Phase.REFERENCE_BINDING_WIREFORMAT, + new HeaderReferenceInterceptor(extensions, + jmsBinding, + jmsResourceFactory, + endpointReference) ); + } + + /** + * Indicates that this binding supports async invocations natively + */ + public boolean supportsNativeAsync() { + return true; + } // end method supportsNativeAsync + +} // end class diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java new file mode 100644 index 0000000000..b96376eb15 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSBindingServiceBindingProvider.java @@ -0,0 +1,265 @@ +/* + * 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.Logger; + +import org.apache.tuscany.sca.assembly.Binding; +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.binding.jms.headers.HeaderServiceInterceptor; +import org.apache.tuscany.sca.binding.jms.host.JMSServiceListener; +import org.apache.tuscany.sca.binding.jms.host.JMSServiceListenerDetails; +import org.apache.tuscany.sca.binding.jms.host.JMSServiceListenerFactory; +import org.apache.tuscany.sca.binding.jms.transport.TransportServiceInterceptor; +import org.apache.tuscany.sca.binding.jms.wire.AsyncResponseDestinationInterceptor; +import org.apache.tuscany.sca.binding.jms.wire.CallbackDestinationInterceptor; +import org.apache.tuscany.sca.binding.jms.wire.OperationPropertiesInterceptor; +import org.apache.tuscany.sca.binding.sca.provider.SCABindingAsyncResponseInvoker; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.core.FactoryExtensionPoint; +import org.apache.tuscany.sca.interfacedef.InterfaceContract; +import org.apache.tuscany.sca.invocation.InvocationChain; +import org.apache.tuscany.sca.invocation.InvokerAsyncResponse; +import org.apache.tuscany.sca.invocation.MessageFactory; +import org.apache.tuscany.sca.invocation.Phase; +import org.apache.tuscany.sca.provider.EndpointAsyncProvider; +import org.apache.tuscany.sca.provider.EndpointProvider; +import org.apache.tuscany.sca.provider.OperationSelectorProvider; +import org.apache.tuscany.sca.provider.OperationSelectorProviderFactory; +import org.apache.tuscany.sca.provider.ProviderFactoryExtensionPoint; +import org.apache.tuscany.sca.provider.WireFormatProvider; +import org.apache.tuscany.sca.provider.WireFormatProviderFactory; +import org.apache.tuscany.sca.runtime.RuntimeComponent; +import org.apache.tuscany.sca.runtime.RuntimeComponentService; +import org.apache.tuscany.sca.runtime.RuntimeEndpoint; + +/** + * Implementation of the JMS service binding provider. + * + * @version $Rev$ $Date$ + */ +public class JMSBindingServiceBindingProvider implements EndpointAsyncProvider, JMSServiceListenerDetails { + private static final Logger logger = Logger.getLogger(JMSBindingServiceBindingProvider.class.getName()); + + private ExtensionPointRegistry registry; + private RuntimeEndpoint endpoint; + private RuntimeComponentService service; + private Binding targetBinding; + private JMSBinding jmsBinding; + private JMSResourceFactory jmsResourceFactory; + private JMSServiceListenerFactory serviceListenerFactory; + private JMSServiceListener serviceListener; + + private RuntimeComponent component; + private InterfaceContract interfaceContract; + + private ProviderFactoryExtensionPoint providerFactories; + private FactoryExtensionPoint modelFactories; + + private MessageFactory messageFactory; + + private OperationSelectorProviderFactory operationSelectorProviderFactory; + private OperationSelectorProvider operationSelectorProvider; + + private WireFormatProviderFactory requestWireFormatProviderFactory; + private WireFormatProvider requestWireFormatProvider; + + private WireFormatProviderFactory responseWireFormatProviderFactory; + private WireFormatProvider responseWireFormatProvider; + + public JMSBindingServiceBindingProvider(ExtensionPointRegistry registry, RuntimeEndpoint endpoint, JMSServiceListenerFactory serviceListenerFactory, ExtensionPointRegistry extensionPoints, JMSResourceFactory jmsResourceFactory) { + this.endpoint = endpoint; + this.component = (RuntimeComponent) endpoint.getComponent(); + this.service = (RuntimeComponentService) endpoint.getService(); + this.jmsBinding = (JMSBinding) endpoint.getBinding(); + this.serviceListenerFactory = serviceListenerFactory; + this.targetBinding = jmsBinding; + this.jmsResourceFactory = jmsResourceFactory; + this.registry = registry; + + if (jmsBinding.getResponseActivationSpecName() != null && jmsBinding.getResponseActivationSpecName().length() > 0) { + throw new JMSBindingException("[BJM30023] response/activationSpec element MUST NOT be present when the binding is being used for an SCA service"); + } + + // Set the default destination when using a connection factory. + // If an activation spec is being used, do not set the destination + // because the activation spec provides the destination. + if (jmsBinding.getDestinationName() == null && + (jmsBinding.getActivationSpecName() == null || jmsBinding.getActivationSpecName().equals(""))) { +// if (!service.isCallback()) { // TODO: 2.x migration, is this check needed? + // use the SCA service name as the default destination name + jmsBinding.setDestinationName(service.getName()); +// } + } + + // Get Message factory + modelFactories = extensionPoints.getExtensionPoint(FactoryExtensionPoint.class); + messageFactory = modelFactories.getFactory(MessageFactory.class); + + // Get the factories/providers for operation selection + this.providerFactories = extensionPoints.getExtensionPoint(ProviderFactoryExtensionPoint.class); + this.operationSelectorProviderFactory = + (OperationSelectorProviderFactory)providerFactories.getProviderFactory(jmsBinding.getOperationSelector().getClass()); + if (this.operationSelectorProviderFactory != null){ + this.operationSelectorProvider = operationSelectorProviderFactory.createServiceOperationSelectorProvider(endpoint); + } + + // Get the factories/providers for wire format + this.requestWireFormatProviderFactory = + (WireFormatProviderFactory)providerFactories.getProviderFactory(jmsBinding.getRequestWireFormat().getClass()); + if (this.requestWireFormatProviderFactory != null){ + this.requestWireFormatProvider = requestWireFormatProviderFactory.createServiceWireFormatProvider(endpoint); + } + + this.responseWireFormatProviderFactory = + (WireFormatProviderFactory)providerFactories.getProviderFactory(jmsBinding.getResponseWireFormat().getClass()); + if (this.responseWireFormatProviderFactory != null){ + this.responseWireFormatProvider = responseWireFormatProviderFactory.createServiceWireFormatProvider(endpoint); + } + + // create an interface contract that reflects both request and response + // wire formats + try { + interfaceContract = (InterfaceContract)service.getInterfaceContract().clone(); + + requestWireFormatProvider.configureWireFormatInterfaceContract(interfaceContract); + responseWireFormatProvider.configureWireFormatInterfaceContract(interfaceContract); + } catch (CloneNotSupportedException ex){ + interfaceContract = service.getInterfaceContract(); + } + } + + public InterfaceContract getBindingInterfaceContract() { + return interfaceContract; + } + + public boolean supportsOneWayInvocation() { + return true; + } + + public void start() { + try { + + this.serviceListener = serviceListenerFactory.createJMSServiceListener(this); + serviceListener.start(); + + } catch (Exception e) { + if (e instanceof JMSBindingException) throw (JMSBindingException)e; + throw new JMSBindingException("Error starting JMSServiceBinding", e); + } + } + + public void stop() { + try { + serviceListener.stop(); + } catch (Exception e) { + if (e instanceof JMSBindingException) throw (JMSBindingException)e; + throw new JMSBindingException("Error stopping JMSServiceBinding", e); + } + } + + public String getDestinationName() { + return serviceListener.getDestinationName(); + } + + /* + * Adds JMS specific interceptors to the binding chain + */ + public void configure() { + + InvocationChain bindingChain = endpoint.getBindingInvocationChain(); + + // add transport interceptor + bindingChain.addInterceptor(Phase.SERVICE_BINDING_TRANSPORT, + new TransportServiceInterceptor(registry, jmsBinding, + jmsResourceFactory, + endpoint) ); + + // add operation selector interceptor + bindingChain.addInterceptor(operationSelectorProvider.getPhase(), + operationSelectorProvider.createInterceptor()); + + // add operationProperties interceptor after operation selector + bindingChain.addInterceptor(Phase.SERVICE_BINDING_OPERATION_SELECTOR, + new OperationPropertiesInterceptor(jmsBinding, endpoint)); + + // add callback destination interceptor after operation selector + bindingChain.addInterceptor(Phase.SERVICE_BINDING_WIREFORMAT, + new CallbackDestinationInterceptor(endpoint)); + + bindingChain.addInterceptor(Phase.SERVICE_BINDING_WIREFORMAT, new HeaderServiceInterceptor(jmsBinding)); + + // add async response interceptor after header interceptor + bindingChain.addInterceptor(Phase.SERVICE_BINDING_WIREFORMAT, + new AsyncResponseDestinationInterceptor(endpoint)); + + // add request wire format + bindingChain.addInterceptor(requestWireFormatProvider.getPhase(), + requestWireFormatProvider.createInterceptor()); + + // add response wire format, but only add it if it's different from the request + if (!jmsBinding.getRequestWireFormat().equals(jmsBinding.getResponseWireFormat())){ + bindingChain.addInterceptor(responseWireFormatProvider.getPhase(), + responseWireFormatProvider.createInterceptor()); + } + + } + + public RuntimeComponent getComponent() { + return component; + } + + public RuntimeComponentService getService() { + return service; + } + + public Binding getTargetBinding() { + return targetBinding; + } + + public JMSBinding getJmsBinding() { + return jmsBinding; + } + + public MessageFactory getMessageFactory() { + return messageFactory; + } + + public JMSResourceFactory getResourceFactory() { + return jmsResourceFactory; + } + + public RuntimeEndpoint getEndpoint() { + return endpoint; + } + + /** + * Indicates that this service binding does support native async service invocations + */ + public boolean supportsNativeAsync() { + return true; + } // end method supportsNativeAsync + + public InvokerAsyncResponse createAsyncResponseInvoker() { + return new JMSBindingAsyncResponseInvoker(null, endpoint); + } // end method createAsyncResponseInvoker + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessor.java new file mode 100644 index 0000000000..821b9d7873 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/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/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessorUtil.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessorUtil.java new file mode 100644 index 0000000000..4b96f23d65 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSMessageProcessorUtil.java @@ -0,0 +1,113 @@ +/* + * 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.Constructor; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; + +/** + * Utility methods to load JMS message processors. + * + * @version $Rev$ $Date$ + */ +public class JMSMessageProcessorUtil { + + /** + * Used to create instances of the JMSResourceFactory and RequestMessageProcessor and ResponseMessageProcessor from + * string based class name provided in the configuration + * + * @param cl ClassLoader + * @param className the string based class name to load and instantiate + * @return the new object + */ + private static Object instantiate(ClassLoader cl, String className, JMSBinding binding) { + Object instance; + if (cl == null) { + cl = binding.getClass().getClassLoader(); + } + + try { + Class<?> clazz; + + try { + clazz = cl.loadClass(className); + } catch (ClassNotFoundException e) { + clazz = binding.getClass().getClassLoader().loadClass(className); + } + + Constructor<?> constructor = clazz.getDeclaredConstructor(new Class[] {JMSBinding.class}); + instance = constructor.newInstance(binding); + + } catch (Throwable e) { + throw new JMSBindingException("Exception instantiating OperationAndDataBinding class", e); + } + + return instance; + } + +// public static JMSMessageProcessor getRequestMessageProcessor(JMSBinding binding) { +// return (JMSMessageProcessor)instantiate(null, binding.getRequestMessageProcessorName(), binding); +// } +// +// public static JMSMessageProcessor getResponseMessageProcessor(JMSBinding binding) { +// return (JMSMessageProcessor)instantiate(null, binding.getResponseMessageProcessorName(), binding); +// } +// + private static Object instantiate(ClassLoader cl, String className, JMSBinding binding, ExtensionPointRegistry registry) { + Object instance; + if (cl == null) { + cl = binding.getClass().getClassLoader(); + } + + try { + Class<?> clazz; + + try { + clazz = cl.loadClass(className); + } catch (ClassNotFoundException e) { + // MJE 07/12/2010 - for OSGi the default message processor belongs to the same bundle as + // this JMSMessageProcessorUtil itself and so the "correct" classloader to use is the classloader + // for THIS class, and not the binding class (which is a different bundle) + // clazz = binding.getClass().getClassLoader().loadClass(className); + clazz = JMSMessageProcessorUtil.class.getClassLoader().loadClass(className); + } + + Constructor<?> constructor = clazz.getDeclaredConstructor(new Class[] {JMSBinding.class, ExtensionPointRegistry.class}); + instance = constructor.newInstance(binding, registry); + + } catch (Throwable e) { + throw new JMSBindingException("Exception instantiating OperationAndDataBinding class", e); + } + + return instance; + } + + public static JMSMessageProcessor getRequestMessageProcessor(ExtensionPointRegistry registry, JMSBinding binding) { + return (JMSMessageProcessor)instantiate(null, binding.getRequestMessageProcessorName(), binding, registry); + } + + public static JMSMessageProcessor getResponseMessageProcessor(ExtensionPointRegistry registry, JMSBinding binding) { + return (JMSMessageProcessor)instantiate(null, binding.getResponseMessageProcessorName(), binding, registry); + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java new file mode 100644 index 0000000000..5aaeca4ebc --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactory.java @@ -0,0 +1,99 @@ +/* + * 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.Connection; +import javax.jms.Destination; +import javax.jms.JMSException; +import javax.jms.Session; +import javax.naming.NamingException; +import javax.resource.spi.ActivationSpec; + +public interface JMSResourceFactory { + + /* + * 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 abstract Connection getConnection() throws NamingException, JMSException; + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.binding.jms.JMSResourceFactory#createSession() + */ + public abstract Session createSession() throws JMSException, NamingException; + + public abstract void closeSession(Session session) throws JMSException; + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.binding.jms.JMSResourceFactory#startConnection() + */ + public abstract void startConnection() throws JMSException, NamingException; + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.binding.jms.JMSResourceFactory#closeConnection() + */ + public abstract void closeConnection() throws JMSException; + + public abstract Destination lookupDestination(String destName) throws NamingException; + + /** + * 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 abstract Destination createDestination(String jndiName) throws NamingException; + + /* + * 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 abstract Connection getResponseConnection() throws NamingException, JMSException; + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.binding.jms.JMSResourceFactory#createSession() + */ + public abstract Session createResponseSession() throws JMSException, NamingException; + + public abstract void closeResponseSession(Session session) throws JMSException; + + public abstract void closeResponseConnection() throws JMSException; + + /* + * Indicates whether connections obtained using getConnection() or getResponseConnection() + * must be closed after each use. This is necessary in environments where connections are + * shared with other users, or where connections cannot be held across transaction boundaries. + */ + public abstract boolean isConnectionClosedAfterUse(); + + public abstract ActivationSpec lookupActivationSpec( + String activationSpecName); +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryExtensionPoint.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryExtensionPoint.java new file mode 100644 index 0000000000..afd2508043 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryExtensionPoint.java @@ -0,0 +1,28 @@ +/* + * 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.JMSBinding; + +public interface JMSResourceFactoryExtensionPoint { + + JMSResourceFactory createJMSResourceFactory(JMSBinding binding); + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryImpl.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryImpl.java new file mode 100644 index 0000000000..7e56fb2329 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/JMSResourceFactoryImpl.java @@ -0,0 +1,359 @@ +/* + * 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 javax.naming.NoInitialContextException; +import javax.resource.spi.ActivationSpec; + +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.extensibility.ClassLoaderContext; +import org.apache.activemq.jndi.ActiveMQInitialContextFactory; + +/** + * Abstracts away any JMS provide specific feature from the JMS binding + * + * @version $Rev$ $Date$ + */ +public class JMSResourceFactoryImpl implements JMSResourceFactory { + + protected String initialContextFactoryName; + protected String connectionFactoryName = "ConnectionFactory"; + protected String jndiURL; + + protected Connection connection; + protected Context context; + protected boolean isConnectionStarted; + private Connection responseConnection; + private String responseConnectionFactoryName; + + public JMSResourceFactoryImpl(String connectionFactoryName, String responseConnectionFactoryName, String initialContextFactoryName, String jndiURL) { + if (connectionFactoryName != null && connectionFactoryName.trim().length() > 0) { + this.connectionFactoryName = connectionFactoryName.trim(); + } + if (responseConnectionFactoryName != null && responseConnectionFactoryName.trim().length() > 0) { + this.responseConnectionFactoryName = responseConnectionFactoryName.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.sca.binding.jms.provider.JMSResourceFactory#getConnection() + */ + public Connection getConnection() throws NamingException, JMSException { + if (connection == null) { + createConnection(); + } + return connection; + } + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#createSession() + */ + public Session createSession() throws JMSException, NamingException { + return getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE); + } + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#closeSession(javax.jms.Session) + */ + public void closeSession(Session session) throws JMSException { + session.close(); + } + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#startConnection() + */ + public void startConnection() throws JMSException, NamingException { + if (!isConnectionStarted) { + getConnection().start(); + isConnectionStarted = true; + } + } + + /* + * (non-Javadoc) + * + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#closeConnection() + */ + public void closeConnection() throws JMSException { + if (connection != null) { + try { + connection.close(); + } catch (JMSException e) { + // if using an embedded broker then when shutting down Tuscany the broker may get closed + // before this stop method is called. I can't see how to detect that so for now just + // ignore the exception if the message is that the transport is already disposed + if (!e.getMessage().contains("disposed")) { + throw e; + } + } + } + } + + protected void createConnection() throws NamingException, JMSException { + Object o = jndiLookUp(connectionFactoryName); + if (o == null) { + throw new JMSBindingException("connection factory not found: " + connectionFactoryName); + } + if (!(o instanceof ConnectionFactory)) { + throw new JMSBindingException("JNDI resource '" + connectionFactoryName +"' is not a JMS ConnectionFactory"); + } + ConnectionFactory connectionFactory = (ConnectionFactory)o; + connection = connectionFactory.createConnection(); + } + + protected 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); + + try { + // Load the JNDI InitialContext (will load the InitialContextFactory, if present) + context = new InitialContext(props); + if( context == null ) { + throw new NamingException(); + } else if ( context.getEnvironment().get(InitialContext.INITIAL_CONTEXT_FACTORY) == null ) { + throw new NamingException(); + } // end if + } catch (NamingException e ) { + context = getInitialContextOsgi( props ); + } // end try + // In the case where the InitialContext fails, check whether performing an OSGi based load succeeds... + + + } + return context; + } // end method getInitialContext + + static final String ACTIVEMQ_FACTORY = "org.apache.activemq.jndi.ActiveMQInitialContextFactory"; + private Context getInitialContextOsgi( Properties props ) throws NamingException { + /** + * For OSGi, need to provide access to the InitialContextFactory for the JMS provider that is going to be used. + * + * The situation is that the InitialContext constructor instantiates an instance of the InitialContextFactory by + * calling "new" using the TCCL - thus there is a need to prepare the TCCL. + * 03/12/2010 MJE - for the present, only worry about ActiveMQ - other providers can be added later + * 10/12/2010 MJE - the following code attempts to get the classloader for the ActiveMQ initial context factory + * it will fail if the ActiveMQ classes are not available in the runtime, but the code will still + * execute (although under OSGi the new InitialContext() operation will fail to find a suitable + * InitialContextFactory object...) + */ + + String contextFactoryName = (String)props.get(Context.INITIAL_CONTEXT_FACTORY); + + ClassLoader ActiveMQCl = null; + try { + if( contextFactoryName == null || ACTIVEMQ_FACTORY.equals(contextFactoryName) ) { + ActiveMQCl = ActiveMQInitialContextFactory.class.getClassLoader(); + props.setProperty(Context.INITIAL_CONTEXT_FACTORY, ACTIVEMQ_FACTORY); + if( props.getProperty(Context.PROVIDER_URL) == null ) { + props.setProperty(Context.PROVIDER_URL, "vm://localhost?broker.persistent=false" ); + } // end if + } // end if + } catch (Exception e) { + // Nothing to do in this case - the ActiveMQCl classloader will simply be null + } // end try + + ClassLoader tccl = ClassLoaderContext.setContextClassLoader(JMSResourceFactoryImpl.class.getClassLoader(), + ActiveMQCl, + Thread.currentThread().getContextClassLoader() ); + try { + // Load the JNDI InitialContext (will load the InitialContextFactory, if present) + return new InitialContext(props); + } finally { + // Restore the TCCL if we changed it + if( tccl != null ) Thread.currentThread().setContextClassLoader(tccl); + } // end try + + } // end method getInitialContextOsgi + + + /** + * 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 + */ + protected 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"); + } + } + } + + /* (non-Javadoc) + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#lookupDestination(java.lang.String) + */ + public Destination lookupDestination(String destName) throws NamingException { + if (destName == null) { + return null; + } + + Destination dest = (Destination)jndiLookUp(destName); + if (dest == null) { + dest = lookupPhysical(destName); + } + return dest; + } + + protected Destination lookupPhysical(String jndiName) { + + // TODO: the SCA JMS spec says a destination name may be a non-jndi plain destination name + +// Session session = null; +// try { +// +// Destination dest; +// session = createSession(); +// dest = session.createQueue(jndiName); +// return dest; +// +// } catch (JMSException e) { +// throw new JMSBindingException(e); +// } catch (NamingException e) { +// throw new JMSBindingException(e); +// } finally { +// if (session != null) { +// try { +// session.close(); +// } catch (JMSException e) { +// throw new JMSBindingException(e); +// } +// } +// } + return null; + } + + /* (non-Javadoc) + * @see org.apache.tuscany.sca.binding.jms.provider.JMSResourceFactory#createDestination(java.lang.String) + */ + public Destination createDestination(String jndiName) throws NamingException { + if ( jndiName == null ) + return null; + + return lookupDestination("dynamicQueues/" + jndiName); + } + + protected Object jndiLookUp(String name) { + Object o = null; + try { + o = getInitialContext().lookup("java:comp/env/" + name); + } catch (Exception ex) { + // ignore + } + if (o == null) { + try { + o = getInitialContext().lookup(name); + } catch (NamingException ex) { + // ignore + } + } + return o; + } + + public Session createResponseSession() throws JMSException, NamingException { + return getResponseConnection().createSession(false, Session.AUTO_ACKNOWLEDGE); + } + + public void closeResponseSession(Session session) throws JMSException { + session.close(); + } + + public Connection getResponseConnection() throws NamingException, JMSException { + if (responseConnection == null) { + if (responseConnectionFactoryName != null) { + ConnectionFactory connectionFactory = (ConnectionFactory)jndiLookUp(responseConnectionFactoryName); + if (connectionFactory == null) { + throw new JMSBindingException("connection factory not found: " + responseConnectionFactoryName); + } + responseConnection = connectionFactory.createConnection(); + } else { + // if no response connection is defined in the SCDL use the request connection + responseConnection = getConnection(); + } + } + return responseConnection; + } + + public void closeResponseConnection() throws JMSException { + if (responseConnection != null && !responseConnection.equals(connection)) { + try { + responseConnection.close(); + } catch (JMSException e) { + // if using an embedded broker then when shutting down Tuscany the broker may get closed + // before this stop method is called. I can't see how to detect that so for now just + // ignore the exception if the message is that the transport is already disposed + if (!e.getMessage().contains("disposed")) { + throw e; + } + } + } + } + + public boolean isConnectionClosedAfterUse() { + // It is assumed this resource factory is used in an environment + // where the connection can be held for the life of the binding. + return false; + } + + public ActivationSpec lookupActivationSpec(String activationSpecName) { + Object o = jndiLookUp(activationSpecName); + if ( o == null ) + return null; + else if (o instanceof ActivationSpec) + return (ActivationSpec) o; + + throw new JMSBindingException("Incorrect resource type for ActivationSpec: " + o.getClass().getName()); + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java new file mode 100644 index 0000000000..db0b94dce1 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/ObjectMessageProcessor.java @@ -0,0 +1,231 @@ +/* + * 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 java.lang.reflect.InvocationTargetException; +import java.util.logging.Logger; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.ObjectMessage; +import javax.jms.Session; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.oasisopen.sca.ServiceRuntimeException; + +/** + * MessageProcessor for sending/receiving Serializable objects with the JMSBinding. + * + */ +public class ObjectMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(ObjectMessageProcessor.class.getName()); + + public ObjectMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + } + + @Override + protected Message createJMSMessage(Session session, Object o) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + + ObjectMessage message = session.createObjectMessage(); + + if (o != null){ + if (!(o instanceof Serializable)) { + throw new IllegalStateException("JMS ObjectMessage payload not Serializable: " + o); + } + + message.setObject((Serializable)o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + @Override + public Object extractPayloadFromJMSMessage(Message msg) { + try { + Object o = ((ObjectMessage)msg).getObject(); + if (o instanceof Throwable ) { + if (o instanceof RuntimeException) { + throw new ServiceRuntimeException("remote service exception, see nested exception", (RuntimeException)o); + } else { + return new InvocationTargetException((Throwable) o); + } + } + } catch (JMSException e) { + throw new JMSBindingException(e); + } + return extractPayload(msg); + } + + @Override + protected Object extractPayload(Message msg) { + try { + + return ((ObjectMessage)msg).getObject(); + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + // special methods for handling operations with single parameters + + public Message createJMSMessageForSingleParamOperation(Session session, Object o, boolean wrapSingleInput) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + + ObjectMessage message = session.createObjectMessage(); + + if (o != null) { + if (!(o instanceof Serializable)) { + throw new IllegalStateException("JMS ObjectMessage payload not Serializable: " + o); + } + + // If the user has specifically requests that single parameters + // be wrapped then leave is as is as it will have already been + // wrapped by Tuscany. Otherwise unwrap it. + if (wrapSingleInput) { + message.setObject((Serializable) o); + } else { // unwrap from array + message.setObject((Serializable) ((Object[]) o)[0]); + } + + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + public Object extractPayloadFromJMSMessageForSingleParamOperation(Message msg, Class<?> argType, boolean wrapSingle) { + // We always have a one arg operation if this method is called so we need to + // decide if the data on the wire is wrapped or not. This is the algorithm. + // + // If the payload is null then create an empty array and pass it on + // If the payload is not an array then it must represent an unwrapped + // single arg. Wrap it up and pass it on + // If the payload is an array then determine if it's a wrapped single arg or not + // If the service interface arg type matches the type of the array and not it's contents + // then it's an unwrapped argument so wrap it and pass it on + // If the service interface arg type matches the type of the contents and not the type + // of the array then the parameter is already wrapped so pass it on as is + // If the service interface arg type matches both the type of the + // array and the type of its contents then assume that the whole array is the + // parameter and decide whether to unwrap it or pass it on as is based on the + // setting of the wrapSingle attribute + // + + try { + Object payload = ((ObjectMessage) msg).getObject(); + + if (payload instanceof Throwable) { + if (payload instanceof RuntimeException) { + throw new ServiceRuntimeException("remote service exception, see nested exception", (RuntimeException) payload); + } else { + return new InvocationTargetException((Throwable) payload); + } + } + + if (payload == null) { + // methodA(null) was not wrapped on wire so wrap it here in order + // that it passes through the rest of the Tuscany wire successfully + return new Object[] { payload }; + } + + boolean payloadIsArray = payload.getClass().isArray(); + + // Non-array payload is single arg + if (!payloadIsArray) { + // methodB(arg) wasn't wrapped on wire so wrap it here in order + // that it passes through the rest of the Tuscany wire successfully + return new Object[] { payload }; + } else { + int size = ((Object[]) payload).length; + + // An initial quick check to determine whether the payload is not + // wrapped. If the array has anything other than a single entry + // then it's not the result of reference side wrapping so wrap it + // here and pass it on + if (size != 1) { + return new Object[] { payload }; + } + + // we know the array has only one entry now so get it + Object arrayContents = ((Object[]) payload)[0]; + + // Is the operation argument the same type as the array itself? + if (argType.isAssignableFrom(payload.getClass())) { + + // So we believe that the whole array is the argument but need + // to check what is in the array to be sure + if (arrayContents == null) { + // There is nothing in the array so it could be an accident that + // the array type matches the argument type, e.g. op(Object) + // so rely on the wrapSingle setting to choose + if (wrapSingle) { + return payload; + } else { + return new Object[] { payload }; + } + } else if (argType.isAssignableFrom(arrayContents.getClass())) { + // We can't tell as the argument type matches both the array type and + // the array contents type so use the wrapSingle setting to choose + if (wrapSingle) { + return payload; + } else { + return new Object[] { payload }; + } + } else { + // So by now we know the whole array is intended to be the + // parameter to wrap it and send it on + return new Object[] { payload }; + } + + } else { + // The array type doesn't match the argument type so assume that the + // array contents will match the argument type and that hence the + // parameter is already wrapped so just send it as is. If the contents + // type doesn't match the argument type a exception will be thrown further + // along the wire + return payload; + } + } + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java new file mode 100644 index 0000000000..0e88b283dc --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/RRBJMSBindingInvoker.java @@ -0,0 +1,357 @@ +/* + * 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.Session; +import javax.naming.NamingException; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.binding.jms.context.JMSBindingContext; +import org.apache.tuscany.sca.core.invocation.InterceptorAsyncImpl; +import org.apache.tuscany.sca.interfacedef.Operation; +import org.apache.tuscany.sca.interfacedef.util.FaultException; +import org.apache.tuscany.sca.invocation.Invoker; +import org.apache.tuscany.sca.invocation.Message; +import org.apache.tuscany.sca.runtime.RuntimeEndpointReference; +import org.oasisopen.sca.ServiceRuntimeException; + +/** + * Invoker for the JMS binding. + * + * @version $Rev$ $Date$ + */ +public class RRBJMSBindingInvoker extends InterceptorAsyncImpl { + + protected Operation operation; + protected String operationName; + + protected JMSBinding jmsBinding; + protected JMSResourceFactory jmsResourceFactory; + protected Destination bindingRequestDest; + protected Destination bindingReplyDest; + protected RuntimeEndpointReference endpointReference; + + public RRBJMSBindingInvoker(Operation operation, JMSResourceFactory jmsResourceFactory, RuntimeEndpointReference epr) { + + this.operation = operation; + operationName = operation.getName(); + + this.endpointReference = epr; + this.jmsBinding = (JMSBinding) epr.getBinding(); + this.jmsResourceFactory = jmsResourceFactory; + + try { + // If this is a callback reference, the destination is determined dynamically based on + // properties of the inbound service request. We should not look for or require a + // statically-configured destination unless a message is received that does not have + // the necessary properties. + bindingRequestDest = lookupDestination(); + bindingReplyDest = lookupResponseDestination(); + } catch (NamingException e) { + throw new JMSBindingException(e); + } // end try + } // end constructor + + /** + * 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) + */ + protected 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) + */ + protected 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 + */ + protected 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 (queueName == null) { + return null; + } + } else { + queueName = jmsBinding.getDestinationName(); + queueType = "JMS Destination "; + qCreateMode = jmsBinding.getDestinationCreate(); + } + + // Remove jms:jndi: prefix if present + if (queueName.startsWith("jms:jndi:")) { + queueName = queueName.substring("jms:jndi:".length()); + } + + 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 + // but don't create when using jms:jndi uri format + if (dest == null && !"jndi".equals(jmsBinding.getDestinationType())) { + 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; + } // end method lookupDestinationQueue + + /** + * Get the next in the chain from the binding invocation chain + */ + public Invoker getNext() { + return (Invoker)endpointReference.getBindingInvocationChain().getHeadInvoker(); + } // end method getNext + + + public org.apache.tuscany.sca.invocation.Message invoke(org.apache.tuscany.sca.invocation.Message tuscanyMsg) { + try { + // populate the message context with JMS binding information + JMSBindingContext context = new JMSBindingContext(); + context.setJmsResourceFactory(jmsResourceFactory); + tuscanyMsg.setBindingContext(context); + + // get a jms session to cover the creation and sending of the message + Session session = context.getJmsSession(); + + context.setRequestDestination(getRequestDestination(tuscanyMsg, session)); + context.setReplyToDestination(getReplyToDestination(session)); + + try { + tuscanyMsg = endpointReference.getBindingInvocationChain().getHeadInvoker().invoke(tuscanyMsg); + } catch (ServiceRuntimeException e) { + if (e.getCause() instanceof InvocationTargetException) { + if ((e.getCause().getCause() instanceof RuntimeException)) { + tuscanyMsg.setFaultBody(e.getCause()); + } else { + tuscanyMsg.setFaultBody(((InvocationTargetException)e.getCause()).getTargetException()); + } + } else if (e.getCause() instanceof FaultException) { + tuscanyMsg.setFaultBody(e.getCause()); + } else { + tuscanyMsg.setFaultBody(e); + } + } catch (IllegalStateException e) { + tuscanyMsg.setFaultBody(e); + } catch (Throwable e) { + tuscanyMsg.setFaultBody(e); + } finally { + context.closeJmsSession(); + if (jmsResourceFactory.isConnectionClosedAfterUse()) { + jmsResourceFactory.closeConnection(); + } + } + + return tuscanyMsg; + } catch (Exception e) { + throw new JMSBindingException(e); + } + } + + protected Destination getRequestDestination(org.apache.tuscany.sca.invocation.Message tuscanyMsg, Session session) throws JMSBindingException, NamingException, JMSException { + Destination requestDestination; +// if (!reference.isCallback()) { // TODO: 2.x migration, is this check needed? +// String toURI = tuscanyMsg.getTo().getURI(); +// if (toURI != null && toURI.startsWith("jms:")) { +// // the msg to uri contains the callback destination name +// // this is an jms physical name not a jndi name so need to use session.createQueue +// requestDestination = session.createQueue(toURI.substring(4)); +// } else { +// requestDestination = lookupDestination(); +// } +// } else { + requestDestination = bindingRequestDest; +// } + + return requestDestination; + } + + protected Destination getReplyToDestination(Session session) throws JMSException, JMSBindingException, NamingException { + Destination replyToDest; + // [rfeng] If the oneway operation is part of bi-directional interface, the JMSReplyTo should be set + if (operation.isNonBlocking() && endpointReference.getComponentReferenceInterfaceContract() + .getCallbackInterface() == null) { + replyToDest = null; + } else { + if (bindingReplyDest != null) { + replyToDest = bindingReplyDest; + } else { + replyToDest = session.createTemporaryQueue(); + } + } + return replyToDest; + } + + /** + * Process forward request message + * @param tuscanyMsg - the request message + * @return the processed version of the request message + */ + public Message processRequest(Message tuscanyMsg) { + try { + // populate the message context with JMS binding information + JMSBindingContext context = new JMSBindingContext(); + context.setJmsResourceFactory(jmsResourceFactory); + tuscanyMsg.setBindingContext(context); + + // get a JMS session to cover the creation and sending of the message + Session session = context.getJmsSession(); + + context.setRequestDestination(getRequestDestination(tuscanyMsg, session)); + context.setReplyToDestination(getReplyToDestination(session)); + + return tuscanyMsg; + } catch (Exception e) { + throw new JMSBindingException(e); + } // end try + } // end method processRequest + + /** + * Post processing for a request message where an error occurred + * @param tuscanyMsg + * @return the post processed message + */ + public Message postProcessRequest(Message tuscanyMsg, Throwable e) { + // Exception handling + if ( e instanceof ServiceRuntimeException ) { + if (e.getCause() instanceof InvocationTargetException) { + if ((e.getCause().getCause() instanceof RuntimeException)) { + tuscanyMsg.setFaultBody(e.getCause()); + } else { + tuscanyMsg.setFaultBody(((InvocationTargetException)e.getCause()).getTargetException()); + } // end if + } else if (e.getCause() instanceof FaultException) { + tuscanyMsg.setFaultBody(e.getCause()); + } else { + tuscanyMsg.setFaultBody(e); + } // end if + } else { + tuscanyMsg.setFaultBody(e); + } // end if + + return postProcessRequest( tuscanyMsg ); + } // end method postProcessRequest + + /** + * General post processing for a request message + * - close out the JMS session & connection + * @param tuscanyMsg + * @return the post processed message + */ + public Message postProcessRequest(Message tuscanyMsg) { + // Close of JMS session + try { + JMSBindingContext context = tuscanyMsg.getBindingContext(); + context.closeJmsSession(); + if (jmsResourceFactory.isConnectionClosedAfterUse()) { + jmsResourceFactory.closeConnection(); + } // end if + } catch (JMSException ex) { + throw new JMSBindingException(ex); + } // end try + return tuscanyMsg; + } // end method postProcessRequest + + /** + * Process response message + * @param tuscanyMsg - the response message + * @return the processed version of the response message + */ + public Message processResponse(Message tuscanyMsg) { + // For async handling, there is nothing to do here + return tuscanyMsg; + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java new file mode 100644 index 0000000000..b468c73e7a --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/TextMessageProcessor.java @@ -0,0 +1,80 @@ +/* + * 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.Logger; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; + +/** + * MessageProcessor for sending/receiving javax.jms.TextMessage with the JMSBinding. + * + * @version $Rev$ $Date$ + */ +public class TextMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(TextMessageProcessor.class.getName()); + + public TextMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + } + + @Override + protected Object extractPayload(Message msg) { + try { + + if (!(msg instanceof TextMessage)) { + throw new IllegalStateException("expecting JMS TextMessage: " + msg); + } + + return ((TextMessage)msg).getText(); + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + @Override + protected Message createJMSMessage(Session session, Object o) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + + TextMessage message = session.createTextMessage(); + + if (o != null){ + message.setText(String.valueOf(o)); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLBytesMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLBytesMessageProcessor.java new file mode 100644 index 0000000000..69a69064d0 --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLBytesMessageProcessor.java @@ -0,0 +1,136 @@ +/* + * 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.IOException; +import java.util.logging.Logger; + +import javax.jms.BytesMessage; +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.common.xml.dom.DOMHelper; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.interfacedef.util.FaultException; +import org.w3c.dom.Node; +import org.xml.sax.SAXException; + +/** + * MessageProcessor for sending/receiving XML javax.jms.BytesMessage with the JMSBinding. + */ +public class XMLBytesMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(XMLBytesMessageProcessor.class.getName()); + + private DOMHelper domHelper; + + public XMLBytesMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + this.domHelper = DOMHelper.getInstance(registry); + } + + @Override + protected Object extractPayload(Message msg) { + try { + + if (!(msg instanceof BytesMessage)) { + throw new IllegalStateException("expecting JMS BytesMessage: " + msg); + } + + long noOfBytes = ((BytesMessage)msg).getBodyLength(); + byte [] bytes = new byte[(int)noOfBytes]; + ((BytesMessage)msg).readBytes(bytes); + ((BytesMessage)msg).reset(); + + Object os; + if (noOfBytes > 0) { + os = domHelper.load(new String(bytes)); + } else { + os = null; + } + return os; + } catch (JMSException e) { + throw new JMSBindingException(e); + } catch (IOException e) { + throw new JMSBindingException(e); + } catch (SAXException e) { + throw new JMSBindingException(e); + } + } + + @Override + public Object extractPayloadFromJMSMessage(Message msg) { + if (msg instanceof BytesMessage) { + return extractPayload(msg); + } else { + return super.extractPayloadFromJMSMessage(msg); + } + } + + @Override + protected Message createJMSMessage(Session session, Object o) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + BytesMessage message = session.createBytesMessage(); + + if (o instanceof Node) { + message.writeBytes(domHelper.saveAsString((Node)o).getBytes()); + } else if ((o instanceof Object[]) && ((Object[])o)[0] instanceof Node) { + message.writeBytes(domHelper.saveAsString((Node)((Object[])o)[0]).getBytes()); + } else if (o != null) { + throw new IllegalStateException("expecting Node payload: " + o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + @Override + public Message createFaultMessage(Session session, Throwable o) { + + if (session == null) { + logger.fine("no response session to create fault message: " + String.valueOf(o)); + return null; + } + if (o instanceof FaultException) { + try { + + BytesMessage message = session.createBytesMessage(); + message.writeBytes(domHelper.saveAsString((Node)((FaultException)o).getFaultInfo()).getBytes()); + message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true); + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } else { + return super.createFaultMessage(session, o); + } + } + +} diff --git a/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java new file mode 100644 index 0000000000..5031f55c6b --- /dev/null +++ b/sandbox/sebastien/java/wrapped/modules/binding-jms-runtime/src/main/java/org/apache/tuscany/sca/binding/jms/provider/XMLTextMessageProcessor.java @@ -0,0 +1,132 @@ +/* + * 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.IOException; +import java.util.logging.Logger; + +import javax.jms.JMSException; +import javax.jms.Message; +import javax.jms.Session; +import javax.jms.TextMessage; + +import org.apache.tuscany.sca.binding.jms.JMSBinding; +import org.apache.tuscany.sca.binding.jms.JMSBindingConstants; +import org.apache.tuscany.sca.binding.jms.JMSBindingException; +import org.apache.tuscany.sca.common.xml.dom.DOMHelper; +import org.apache.tuscany.sca.core.ExtensionPointRegistry; +import org.apache.tuscany.sca.interfacedef.util.FaultException; +import org.w3c.dom.Element; +import org.w3c.dom.Node; +import org.xml.sax.SAXException; + +/** + * MessageProcessor for sending/receiving XML javax.jms.TextMessage with the JMSBinding. + * + * @version $Rev$ $Date$ + */ +public class XMLTextMessageProcessor extends AbstractMessageProcessor { + private static final Logger logger = Logger.getLogger(XMLTextMessageProcessor.class.getName()); + + private DOMHelper domHelper; + + public XMLTextMessageProcessor(JMSBinding jmsBinding, ExtensionPointRegistry registry) { + super(jmsBinding); + this.domHelper = DOMHelper.getInstance(registry); + } + + @Override + protected Object extractPayload(Message msg) { + try { + + String xml = ((TextMessage)msg).getText(); + Object os; + if (xml != null) { + os = domHelper.load(xml); + } else { + os = null; + } + return os; + + } catch (IOException e) { + throw new JMSBindingException(e); + } catch (JMSException e) { + throw new JMSBindingException(e); + } catch (SAXException e) { + throw new JMSBindingException(e); + } + } + + @Override + public Object extractPayloadFromJMSMessage(Message msg) { + if (msg instanceof TextMessage) { + return extractPayload(msg); + } else { + return super.extractPayloadFromJMSMessage(msg); + } + } + + @Override + protected Message createJMSMessage(Session session, Object o) { + if (session == null) { + logger.fine("no response session to create message: " + String.valueOf(o)); + return null; + } + try { + + TextMessage message = session.createTextMessage(); + + if (o instanceof Element) { + message.setText(domHelper.saveAsString((Node)o)); + } else if ((o instanceof Object[]) && ((Object[])o)[0] instanceof Node) { + message.setText(domHelper.saveAsString((Node)((Object[])o)[0])); + } else if (o != null) { + throw new IllegalStateException("expecting Node payload: " + o); + } + + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } + + @Override + public Message createFaultMessage(Session session, Throwable o) { + if (session == null) { + logger.fine("no response session to create fault message: " + String.valueOf(o)); + return null; + } + if (o instanceof FaultException) { + try { + + TextMessage message = session.createTextMessage(); + message.setText(domHelper.saveAsString((Node)((FaultException)o).getFaultInfo())); + message.setBooleanProperty(JMSBindingConstants.FAULT_PROPERTY, true); + return message; + + } catch (JMSException e) { + throw new JMSBindingException(e); + } + } else { + return super.createFaultMessage(session, o); + } + } + +} |