diff options
author | antelder <antelder@13f79535-47bb-0310-9956-ffa450edef68> | 2009-01-27 15:31:53 +0000 |
---|---|---|
committer | antelder <antelder@13f79535-47bb-0310-9956-ffa450edef68> | 2009-01-27 15:31:53 +0000 |
commit | 1206dc69a267f18b68cb086e06dcc5f3f8ce9987 (patch) | |
tree | 7a53ea3d3941527ce3f682e054f79bb42d1dec3f /branches/sca-java-1.x/modules/host-jms-asf/src/main | |
parent | ccfa9ff9449b1052e344eee6a7de12d1454db051 (diff) |
Start to add some JMS host default implementations
git-svn-id: http://svn.us.apache.org/repos/asf/tuscany@738112 13f79535-47bb-0310-9956-ffa450edef68
Diffstat (limited to 'branches/sca-java-1.x/modules/host-jms-asf/src/main')
2 files changed, 456 insertions, 0 deletions
diff --git a/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/ASFListener.java b/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/ASFListener.java new file mode 100644 index 0000000000..ad48468a95 --- /dev/null +++ b/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/ASFListener.java @@ -0,0 +1,233 @@ +/*
+ * 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.host.jms.asf;
+
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageConsumer;
+import javax.jms.MessageListener;
+import javax.jms.Queue;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBinding;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+import org.apache.tuscany.sca.host.jms.JMSServiceListener;
+import org.apache.tuscany.sca.work.WorkScheduler;
+
+/**
+ * Implementation of the JMS service binding provider.
+ *
+ * @version $Rev: 721811 $ $Date: 2008-11-30 13:46:51 +0000 (Sun, 30 Nov 2008) $
+ */
+public class ASFListener implements JMSServiceListener {
+ private static final Logger logger = Logger.getLogger(ASFListener.class.getName());
+
+ private MessageListener listener;
+ private String serviceName;
+ private boolean isCallbackService;
+ private JMSBinding jmsBinding;
+ private WorkScheduler workScheduler;
+
+ private JMSResourceFactoryImpl jmsResourceFactory;
+ private MessageConsumer consumer;
+ private boolean running;
+
+ private Destination destination;
+
+ public ASFListener(MessageListener listener, String serviceName, boolean isCallbackService, JMSBinding jmsBinding, WorkScheduler workScheduler) {
+ this.listener = listener;
+ this.serviceName = serviceName;
+ this.isCallbackService = isCallbackService;
+ this.jmsBinding = jmsBinding;
+ this.workScheduler = workScheduler;
+ this.jmsResourceFactory = new JMSResourceFactoryImpl(jmsBinding.getConnectionFactoryName(), jmsBinding.getInitialContextFactoryName(), jmsBinding.getJndiURL());
+ }
+
+ public void start() {
+ this.running = true;
+
+ try {
+ registerListerner();
+ } catch (Exception e) {
+ throw new JMSBindingException("Error starting JMSServiceBinding", e);
+ }
+ }
+
+ public void stop() {
+ this.running = false;
+ try {
+ consumer.close();
+ jmsResourceFactory.closeConnection();
+ } catch (Exception e) {
+ // 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 (!"Transport disposed.".equals(e.getMessage())) {
+ throw new JMSBindingException("Error stopping JMSServiceBinding", e);
+ }
+ }
+ }
+
+ private void registerListerner() throws NamingException, JMSException {
+
+ Session session = jmsResourceFactory.createSession();
+ destination = lookupDestinationQueue();
+ if (destination == null) {
+ destination = session.createTemporaryQueue();
+ }
+
+ if (jmsBinding.getJMSSelector() != null) {
+ consumer = session.createConsumer(destination, jmsBinding.getJMSSelector());
+ } else {
+ consumer = session.createConsumer(destination);
+ }
+
+ try {
+
+ consumer.setMessageListener(listener);
+ jmsResourceFactory.startConnection();
+
+ } catch (javax.jms.IllegalStateException e) {
+
+ // setMessageListener not allowed in JEE container so use Tuscany threads
+
+ jmsResourceFactory.startConnection();
+ workScheduler.scheduleWork(new Runnable() {
+ public void run() {
+ try {
+ while (running) {
+ final Message msg = consumer.receive();
+ workScheduler.scheduleWork(new Runnable() {
+ public void run() {
+ try {
+ listener.onMessage(msg);
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ });
+ }
+ } catch (Exception e) {
+ e.printStackTrace();
+ }
+ }
+ });
+ }
+ logger.log(Level.INFO, "JMS " + (isCallbackService ? "callback service" : "service")
+ + " '"
+ + serviceName
+ + "' listening on destination "
+ + ((destination instanceof Queue) ? ((Queue)destination).getQueueName() : ((Topic)destination).getTopicName()));
+ }
+
+ /**
+ * Looks up the Destination Queue for the JMS Binding.
+ * <p>
+ * What happens in the look up will depend on the create mode specified for the JMS Binding:
+ * <ul>
+ * <li>always - the JMS queue is always created. It is an error if the queue already exists
+ * <li>ifnotexist - the JMS queue is created if it does not exist. It is not an error if the queue already exists
+ * <li>never - the JMS queue is never created. It is an error if the queue does not exist
+ * </ul>
+ * See the SCA JMS Binding specification for more information.
+ * <p>
+ *
+ * @return The Destination queue.
+ * @throws NamingException Failed to lookup JMS queue
+ * @throws JMSBindingException Failed to lookup JMS Queue. Probable cause is that the JMS queue's current existence/non-existence is not
+ * compatible with the create mode specified on the binding
+ */
+ private Destination lookupDestinationQueue() throws NamingException, JMSBindingException {
+
+ if (isCallbackService && JMSBindingConstants.DEFAULT_DESTINATION_NAME.equals(jmsBinding.getDestinationName())) {
+ // if its a callback service returning null indicates to use a temporary queue
+ return null;
+ }
+
+ Destination destination = jmsResourceFactory.lookupDestination(jmsBinding.getDestinationName());
+
+ String qCreateMode = jmsBinding.getDestinationCreate();
+ if (qCreateMode.equals(JMSBindingConstants.CREATE_ALWAYS)) {
+ // In this mode, the queue must not already exist as we are creating it
+ if (destination != null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " already exists but has create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + serviceName
+ + " listener");
+ }
+
+ // Create the queue
+ destination = jmsResourceFactory.createDestination(jmsBinding.getDestinationName());
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_IF_NOT_EXIST)) {
+ // In this mode, the queue may nor may not exist. It will be created if it does not exist
+ if (destination == null) {
+ destination = jmsResourceFactory.createDestination(jmsBinding.getDestinationName());
+ }
+
+ } else if (qCreateMode.equals(JMSBindingConstants.CREATE_NEVER)) {
+ // In this mode, the queue must have already been created.
+ if (destination == null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " not found but create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + serviceName
+ + " listener");
+ }
+ }
+
+ // Make sure we ended up with a queue
+ if (destination == null) {
+ throw new JMSBindingException("JMS Destination " + jmsBinding.getDestinationName()
+ + " not found with create mode of \""
+ + qCreateMode
+ + "\" while registering service "
+ + serviceName
+ + " listener");
+ }
+
+ return destination;
+ }
+
+ public String getDestinationName() {
+ try {
+ if (destination instanceof Queue) {
+ return ((Queue)destination).getQueueName();
+ } else if (destination instanceof Topic) {
+ return ((Topic)destination).getTopicName();
+ } else {
+ return null;
+ }
+ } catch (JMSException e) {
+ throw new JMSBindingException(e);
+ }
+ }
+
+}
diff --git a/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/JMSResourceFactoryImpl.java b/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/JMSResourceFactoryImpl.java new file mode 100644 index 0000000000..fc327d593d --- /dev/null +++ b/branches/sca-java-1.x/modules/host-jms-asf/src/main/java/org/apache/tuscany/sca/host/jms/asf/JMSResourceFactoryImpl.java @@ -0,0 +1,223 @@ +/*
+ * 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.host.jms.asf;
+
+import java.util.Properties;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+import javax.jms.Destination;
+import javax.jms.JMSException;
+import javax.jms.Session;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingConstants;
+import org.apache.tuscany.sca.binding.jms.impl.JMSBindingException;
+
+/**
+ * Abstracts away any JMS provide specific feature from the JMS binding
+ *
+ * @version $Rev: 705229 $ $Date: 2008-10-16 13:43:47 +0100 (Thu, 16 Oct 2008) $
+ */
+public class JMSResourceFactoryImpl {
+
+ protected String initialContextFactoryName;
+ protected String connectionFactoryName = "ConnectionFactory";
+ protected String jndiURL;
+
+ protected Connection connection;
+ protected Context context;
+ protected boolean isConnectionStarted;
+
+ public JMSResourceFactoryImpl(String connectionFactoryName, String initialContextFactoryName, String jndiURL) {
+ if (connectionFactoryName != null && connectionFactoryName.trim().length() > 0) {
+ this.connectionFactoryName = connectionFactoryName.trim();
+ }
+ if (initialContextFactoryName != null && initialContextFactoryName.trim().length() > 0) {
+ this.initialContextFactoryName = initialContextFactoryName.trim();
+ }
+ if (jndiURL != null) {
+ this.jndiURL = jndiURL.trim();
+ }
+ }
+
+ /*
+ * This is a simple implementation where a connection is created per binding Ideally the resource factory should be
+ * able to leverage the host environment to provide connection pooling if it can. E.g. if Tuscany is running inside
+ * an AppServer Then we could leverage the JMS resources it provides
+ *
+ * @see org.apache.tuscany.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#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 {
+ ConnectionFactory connectionFactory = (ConnectionFactory)jndiLookUp(connectionFactoryName);
+ if (connectionFactory == null) {
+ throw new JMSBindingException("connection factory not found: " + connectionFactoryName);
+ }
+ 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);
+
+ context = new InitialContext(props);
+ }
+ return context;
+ }
+
+ /**
+ * If using the WAS JMS Client with a non-IBM JRE then an additional
+ * environment property needs to be set to initialize the ORB correctly.
+ * See: http://www-1.ibm.com/support/docview.wss?uid=swg24012804
+ */
+ 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 (JMSBindingConstants.DEFAULT_DESTINATION_NAME.equals(destName)) {
+ 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 {
+ 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;
+ }
+}
|