openjpa-kernel/src/main/java/org/apache/openjpa/event/JMSRemoteCommitProvider.java - openjpa - Git at Google

 /*
  * 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.openjpa.event;

 import java.util.Properties;

 import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.Message;
 import javax.jms.MessageListener;
 import javax.jms.ObjectMessage;
 import javax.jms.Session;
 import javax.jms.Topic;
 import javax.jms.TopicConnection;
 import javax.jms.TopicConnectionFactory;
 import javax.jms.TopicPublisher;
 import javax.jms.TopicSession;
 import javax.jms.TopicSubscriber;
 import javax.naming.Context;
 import javax.naming.InitialContext;
 import javax.naming.NamingException;

 import org.apache.openjpa.lib.conf.Configurable;
 import org.apache.openjpa.lib.conf.GenericConfigurable;
 import org.apache.openjpa.lib.util.Localizer;
 import org.apache.openjpa.lib.util.Options;
 import org.apache.openjpa.util.OpenJPAException;
 import org.apache.openjpa.util.UserException;

 /**
  * JMS-based implementation of {@link RemoteCommitProvider} that
  * listens for object modifications and propagates those changes to
  * other RemoteCommitProviders over a JMS topic.
  *
  * @author Patrick Linskey
  * @since 0.2.5.0
  */
 public class JMSRemoteCommitProvider
     extends AbstractRemoteCommitProvider
     implements Configurable, GenericConfigurable, ExceptionListener {

     private static Localizer s_loc = Localizer.forPackage
         (JMSRemoteCommitProvider.class);

     private String _topicName = "topic/OpenJPACommitProviderTopic";
     private String _tcfName = "java:/ConnectionFactory";
     private Properties _ctxProps = null;
     private int _reconnectAttempts = 0;
     private TopicConnection _connection;
     private TopicSession _session;
     private TopicPublisher _publisher;
     private ClassLoader _appClassLoader;

     /**
      * Sets the JMS Topic name. Defaults to
      * <code>topic/OpenJPACommitProviderTopic</code>.
      */
     public void setTopic(String name) {
         _topicName = name;
     }

     /**
      * Sets the JMS TopicConnectionFactory name. Defaults to
      * <code>java:/ConnectionFactory</code>.
      */
     public void setTopicConnectionFactory(String name) {
         _tcfName = name;
     }

     /**
      * The number of times to attempt to reconnect after a JMS send exception
      * is detected. Defaults to 0, meaning no attempt to reconnect is made;
      * the exception is logged and ignored.
      */
     public void setExceptionReconnectAttempts(int attempts) {
         _reconnectAttempts = attempts;
     }

     /**
      * Set a map of properties to pass to the {@link InitialContext}
      * constructor for JNDI lookups. Implementation of
      * {@link GenericConfigurable}.
      */
     @Override
     public void setInto(Options opts) {
         if (opts != null && !opts.isEmpty()) {
             _ctxProps = new Properties();
             _ctxProps.putAll(opts);
             opts.clear();
         } else
             _ctxProps = null;
     }

     /**
      * Returns a new {@link Context} object for use by this provider.
      */
     protected Context newContext()
         throws NamingException {
         if (_ctxProps == null)
             return new InitialContext();
         return new InitialContext(_ctxProps);
     }

     // ---------- RemoteCommitProvider implementation ----------

     @Override
     public void broadcast(RemoteCommitEvent event) {
         try {
             _publisher.publish(createMessage(event));
             if (log.isTraceEnabled())
                 log.trace(s_loc.get("jms-sent-update", _topicName));
         } catch (JMSException jmse) {
             if (log.isWarnEnabled())
                 log.warn(s_loc.get("jms-send-error", _topicName), jmse);
         }
     }

     @Override
     public void close() {
         try {
             if (_connection != null) {
                 _connection.close();
                 if (log.isInfoEnabled())
                     log.info(s_loc.get("jms-close-listener", _topicName));
             }
         }
         catch (JMSException jmse) {
             if (log.isWarnEnabled())
                 log.warn(s_loc.get("jms-close-error", _topicName), jmse);
         } finally {
             _connection = null;
         }
     }

     // ---------- Configurable implementation ----------

     /**
      * Subclasses that need to perform actions in
      * {@link Configurable#endConfiguration} must invoke this method.
      */
     @Override
     public void endConfiguration() {
         super.endConfiguration();
         _appClassLoader = Thread.currentThread().getContextClassLoader();
         connect();
     }

     protected void connect() {
         try {
             Context ctx = newContext();
             TopicConnectionFactory tcf =
                 (TopicConnectionFactory) ctx.lookup(_tcfName);
             Topic topic = (Topic) ctx.lookup(_topicName);
             ctx.close();

             _connection = tcf.createTopicConnection();

             // false == not transacted.
             _session = _connection.createTopicSession
                 (false, Session.AUTO_ACKNOWLEDGE);

             // create a publisher
             _publisher = _session.createPublisher(topic);

             // create a subscriber.
             TopicSubscriber s = _session.createSubscriber(topic, null,
                 /* noLocal: */ true);

             MessageListener l = getMessageListener();
             s.setMessageListener(l);
             _connection.start();
             _connection.setExceptionListener(this);
             if (log.isInfoEnabled())
                 log.info(s_loc.get("jms-start-listener", _topicName));
         } catch (OpenJPAException ke) {
             throw ke;
         } catch (Exception e) {
             throw new UserException(s_loc.get("jms-provider-config",
                 _topicName, _tcfName), e).setFatal(true);
         }
     }

     /* *
      * Returns a {@link javax.jms.MessageListener} capable of
      * understanding and processing messages created by {@link #createMessage}.
      *  The listener returned by this method is responsible for
      * notifying the provider that a remote event has been received.
      */
     protected MessageListener getMessageListener() {
         return new MessageListener() {
             @Override
             public void onMessage(Message m) {
                 if (!(m instanceof ObjectMessage)) {
                     if (log.isWarnEnabled())
                         log.warn(s_loc.get("jms-receive-error-3",
                             _topicName, m.getClass().getName()));
                     return;
                 }

                 ClassLoader saveCls = Thread.currentThread()
                     .getContextClassLoader();
                 try {
                     if (saveCls != _appClassLoader)
                         Thread.currentThread().setContextClassLoader(
                             _appClassLoader);
                     ObjectMessage om = (ObjectMessage) m;
                     Object o = om.getObject();

                     if (o instanceof RemoteCommitEvent) {
                     	if (log.isTraceEnabled())
                     		log.trace(s_loc.get("jms-received-update",
                     				_topicName));

                     	RemoteCommitEvent rce = (RemoteCommitEvent) o;
                     	fireEvent(rce);
                     } else {
                     	if (log.isWarnEnabled())
                     		log.warn(s_loc.get("jms-receive-error-2",
                                     o.getClass().getName(), _topicName));
                     }
                 } catch (JMSException jmse) {
                     if (log.isWarnEnabled())
                         log.warn(s_loc.get("jms-receive-error-1"), jmse);
                 } finally {
                     if (saveCls != _appClassLoader)
                         Thread.currentThread().setContextClassLoader(saveCls);
                 }

             }
         };
     }

     /**
      * Returns a new {@link Message} to send to the topic. This
      * implementation creates an {@link ObjectMessage}.
      */
     protected Message createMessage(RemoteCommitEvent event)
         throws JMSException {
         return _session.createObjectMessage(event);
     }

     @Override
     public void onException(JMSException ex) {
         if (log.isWarnEnabled())
             log.warn(s_loc.get("jms-listener-error", _topicName), ex);
         if (_reconnectAttempts <= 0)
             return;

         close();
         boolean connected = false;
         for (int i = 0; !connected && i < _reconnectAttempts; i++) {
             try {
                 if (log.isInfoEnabled())
                     log.info(s_loc.get("jms-reconnect-attempt", _topicName,
                         String.valueOf(i + 1)));
                 connect();
                 connected = true;
             } catch (Exception e) {
                 if (log.isInfoEnabled())
                     log.info(s_loc.get("jms-reconnect-fail", _topicName), e);
                 try {
                     Thread.sleep(1000);
                 } catch (InterruptedException ie) {
                     break;
                 }
             }
         }

         if (!connected && log.isErrorEnabled())
             log.error(s_loc.get("jms-cant-reconnect", _topicName,
                 String.valueOf(_reconnectAttempts)));
         else if (connected && log.isInfoEnabled())
             log.info(s_loc.get("jms-reconnected", _topicName));
 	}
 }
	/*
	* 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.openjpa.event;

	import java.util.Properties;

	import javax.jms.ExceptionListener;
	import javax.jms.JMSException;
	import javax.jms.Message;
	import javax.jms.MessageListener;
	import javax.jms.ObjectMessage;
	import javax.jms.Session;
	import javax.jms.Topic;
	import javax.jms.TopicConnection;
	import javax.jms.TopicConnectionFactory;
	import javax.jms.TopicPublisher;
	import javax.jms.TopicSession;
	import javax.jms.TopicSubscriber;
	import javax.naming.Context;
	import javax.naming.InitialContext;
	import javax.naming.NamingException;

	import org.apache.openjpa.lib.conf.Configurable;
	import org.apache.openjpa.lib.conf.GenericConfigurable;
	import org.apache.openjpa.lib.util.Localizer;
	import org.apache.openjpa.lib.util.Options;
	import org.apache.openjpa.util.OpenJPAException;
	import org.apache.openjpa.util.UserException;

	/**
	* JMS-based implementation of {@link RemoteCommitProvider} that
	* listens for object modifications and propagates those changes to
	* other RemoteCommitProviders over a JMS topic.
	*
	* @author Patrick Linskey
	* @since 0.2.5.0
	*/
	public class JMSRemoteCommitProvider
	extends AbstractRemoteCommitProvider
	implements Configurable, GenericConfigurable, ExceptionListener {

	private static Localizer s_loc = Localizer.forPackage
	(JMSRemoteCommitProvider.class);

	private String _topicName = "topic/OpenJPACommitProviderTopic";
	private String _tcfName = "java:/ConnectionFactory";
	private Properties _ctxProps = null;
	private int _reconnectAttempts = 0;
	private TopicConnection _connection;
	private TopicSession _session;
	private TopicPublisher _publisher;
	private ClassLoader _appClassLoader;

	/**
	* Sets the JMS Topic name. Defaults to
	* <code>topic/OpenJPACommitProviderTopic</code>.
	*/
	public void setTopic(String name) {
	_topicName = name;
	}

	/**
	* Sets the JMS TopicConnectionFactory name. Defaults to
	* <code>java:/ConnectionFactory</code>.
	*/
	public void setTopicConnectionFactory(String name) {
	_tcfName = name;
	}

	/**
	* The number of times to attempt to reconnect after a JMS send exception
	* is detected. Defaults to 0, meaning no attempt to reconnect is made;
	* the exception is logged and ignored.
	*/
	public void setExceptionReconnectAttempts(int attempts) {
	_reconnectAttempts = attempts;
	}

	/**
	* Set a map of properties to pass to the {@link InitialContext}
	* constructor for JNDI lookups. Implementation of
	* {@link GenericConfigurable}.
	*/
	@Override
	public void setInto(Options opts) {
	if (opts != null && !opts.isEmpty()) {
	_ctxProps = new Properties();
	_ctxProps.putAll(opts);
	opts.clear();
	} else
	_ctxProps = null;
	}

	/**
	* Returns a new {@link Context} object for use by this provider.
	*/
	protected Context newContext()
	throws NamingException {
	if (_ctxProps == null)
	return new InitialContext();
	return new InitialContext(_ctxProps);
	}

	// ---------- RemoteCommitProvider implementation ----------

	@Override
	public void broadcast(RemoteCommitEvent event) {
	try {
	_publisher.publish(createMessage(event));
	if (log.isTraceEnabled())
	log.trace(s_loc.get("jms-sent-update", _topicName));
	} catch (JMSException jmse) {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-send-error", _topicName), jmse);
	}
	}

	@Override
	public void close() {
	try {
	if (_connection != null) {
	_connection.close();
	if (log.isInfoEnabled())
	log.info(s_loc.get("jms-close-listener", _topicName));
	}
	}
	catch (JMSException jmse) {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-close-error", _topicName), jmse);
	} finally {
	_connection = null;
	}
	}

	// ---------- Configurable implementation ----------

	/**
	* Subclasses that need to perform actions in
	* {@link Configurable#endConfiguration} must invoke this method.
	*/
	@Override
	public void endConfiguration() {
	super.endConfiguration();
	_appClassLoader = Thread.currentThread().getContextClassLoader();
	connect();
	}

	protected void connect() {
	try {
	Context ctx = newContext();
	TopicConnectionFactory tcf =
	(TopicConnectionFactory) ctx.lookup(_tcfName);
	Topic topic = (Topic) ctx.lookup(_topicName);
	ctx.close();

	_connection = tcf.createTopicConnection();

	// false == not transacted.
	_session = _connection.createTopicSession
	(false, Session.AUTO_ACKNOWLEDGE);

	// create a publisher
	_publisher = _session.createPublisher(topic);

	// create a subscriber.
	TopicSubscriber s = _session.createSubscriber(topic, null,
	/* noLocal: */ true);

	MessageListener l = getMessageListener();
	s.setMessageListener(l);
	_connection.start();
	_connection.setExceptionListener(this);
	if (log.isInfoEnabled())
	log.info(s_loc.get("jms-start-listener", _topicName));
	} catch (OpenJPAException ke) {
	throw ke;
	} catch (Exception e) {
	throw new UserException(s_loc.get("jms-provider-config",
	_topicName, _tcfName), e).setFatal(true);
	}
	}

	/* *
	* Returns a {@link javax.jms.MessageListener} capable of
	* understanding and processing messages created by {@link #createMessage}.
	* The listener returned by this method is responsible for
	* notifying the provider that a remote event has been received.
	*/
	protected MessageListener getMessageListener() {
	return new MessageListener() {
	@Override
	public void onMessage(Message m) {
	if (!(m instanceof ObjectMessage)) {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-receive-error-3",
	_topicName, m.getClass().getName()));
	return;
	}

	ClassLoader saveCls = Thread.currentThread()
	.getContextClassLoader();
	try {
	if (saveCls != _appClassLoader)
	Thread.currentThread().setContextClassLoader(
	_appClassLoader);
	ObjectMessage om = (ObjectMessage) m;
	Object o = om.getObject();

	if (o instanceof RemoteCommitEvent) {
	if (log.isTraceEnabled())
	log.trace(s_loc.get("jms-received-update",
	_topicName));

	RemoteCommitEvent rce = (RemoteCommitEvent) o;
	fireEvent(rce);
	} else {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-receive-error-2",
	o.getClass().getName(), _topicName));
	}
	} catch (JMSException jmse) {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-receive-error-1"), jmse);
	} finally {
	if (saveCls != _appClassLoader)
	Thread.currentThread().setContextClassLoader(saveCls);
	}

	}
	};
	}

	/**
	* Returns a new {@link Message} to send to the topic. This
	* implementation creates an {@link ObjectMessage}.
	*/
	protected Message createMessage(RemoteCommitEvent event)
	throws JMSException {
	return _session.createObjectMessage(event);
	}

	@Override
	public void onException(JMSException ex) {
	if (log.isWarnEnabled())
	log.warn(s_loc.get("jms-listener-error", _topicName), ex);
	if (_reconnectAttempts <= 0)
	return;

	close();
	boolean connected = false;
	for (int i = 0; !connected && i < _reconnectAttempts; i++) {
	try {
	if (log.isInfoEnabled())
	log.info(s_loc.get("jms-reconnect-attempt", _topicName,
	String.valueOf(i + 1)));
	connect();
	connected = true;
	} catch (Exception e) {
	if (log.isInfoEnabled())
	log.info(s_loc.get("jms-reconnect-fail", _topicName), e);
	try {
	Thread.sleep(1000);
	} catch (InterruptedException ie) {
	break;
	}
	}
	}

	if (!connected && log.isErrorEnabled())
	log.error(s_loc.get("jms-cant-reconnect", _topicName,
	String.valueOf(_reconnectAttempts)));
	else if (connected && log.isInfoEnabled())
	log.info(s_loc.get("jms-reconnected", _topicName));
	}
	}