public JMSCacheInvalidationBridge() 
{
   // Static --------------------------------------------------------
   // Constructors --------------------------------------------------
       super ();
}

 public  void batchInvalidate(BatchInvalidation[] invalidations,
    boolean asynchronous) 
{
      if ( (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
            && this.publishingAuthorized)
      {         
         JMSCacheInvalidationMessage msg = new JMSCacheInvalidationMessage (this.serviceId, invalidations);
         this.sendJMSInvalidationEvent (msg);
      }
}

 public int getAcknowledgeMode() 
{
      return this.acknowledgeMode;
}

 public String getConnectionFactoryName() 
{
      return this.connectionFactoryName;
}

 protected InitialContext getInitialContext() throws NamingException 
{
      if (providerUrl == null) 
      {
         return new InitialContext();
      }
      else
      {
         log.debug("Using Context.PROVIDER_URL: " + providerUrl);
         java.util.Properties props = new java.util.Properties(System.getProperties());
         props.put(Context.PROVIDER_URL, providerUrl);
         return new InitialContext(props);
      }
}

 public String getInvalidationManager() 
{
      return this.invalidationManagerName;
}

 public int getPropagationMode() 
{
      return this.propagationMode;
}

 public String getProviderUrl() 
{
      return providerUrl;
}

 protected synchronized TopicPublisher getPublisher() 
{
      return this.pub;
}

 protected synchronized TopicSession getSession() 
{
      return this.session;
}

 public String getTopicName() 
{
      return this.topicName;
}

 public  void groupIsDropped(String groupInvalidationName) 
{
      // we don't manage groups dynamically, so we don't really care...
      //
}

 public  void invalidate(String invalidationGroupName,
    Serializable[] keys,
    boolean asynchronous) 
{
      if ( (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
            && this.publishingAuthorized)
      {         
         JMSCacheInvalidationMessage msg = new JMSCacheInvalidationMessage (
                  this.serviceId, 
                  invalidationGroupName, 
                  keys);
         this.sendJMSInvalidationEvent (msg);
      }
}

 public  void invalidate(String invalidationGroupName,
    Serializable key,
    boolean asynchronous) 
{
      if ( (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
            && this.publishingAuthorized)
      {         
         JMSCacheInvalidationMessage msg = new JMSCacheInvalidationMessage (
                  this.serviceId, 
                  invalidationGroupName, 
                  new Serializable[] {key} );
         this.sendJMSInvalidationEvent (msg);
      }
}

 public  void invalidateAll(String groupName,
    boolean asynchronous) 
{
      if ( (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
            && this.publishingAuthorized)
      {
         JMSCacheInvalidationMessage msg = new JMSCacheInvalidationMessage(
                  this.serviceId,
                  groupName
         );
         this.sendJMSInvalidationEvent (msg);
      }
}

 public boolean isTransacted() 
{
      return this.transacted;
}

 public  void newGroupCreated(String groupInvalidationName) 
{
      // we don't manage groups dynamically, so we don't really care...
      //
}

 public  void onMessage(Message msg) 
{
      // just to make sure we are in the good mode
      //
      if (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_ONLY_BRIDGE_PROPAGATION)
      {         
         try
         {
            ObjectMessage objmsg = (ObjectMessage)msg;
            if (!objmsg.getJMSType().equals(JMS_CACHE_INVALIDATION_BRIDGE)) return;          
            JMSCacheInvalidationMessage content = (JMSCacheInvalidationMessage)objmsg.getObject();
            // Not very efficient as the whole message must be unserialized just to check
            // if we were the emitter. Maybe wrapping this in a byte array would be more efficient
            //
            if (!content.emitter.equals (this.serviceId))
            {
               if(content.invalidateAllGroupName != null)
               {
                  invalidationSubscription.invalidateAll(content.invalidateAllGroupName);
               }
               else
               {
                  invalidationSubscription.batchInvalidate (content.getInvalidations ());
               }
            }
         }
         catch (Exception ex)
         {
            log.warn(ex.getMessage());
         }
      }
}

 protected  void sendJMSInvalidationEvent(JMSCacheInvalidationMessage invalidationMsg) 
{
      try
      {
         if (log.isTraceEnabled ())
            log.trace("sending JMS message for cache invalidation" + invalidationMsg);
         try
         {
            ObjectMessage msg = getSession().createObjectMessage();
            msg.setJMSType(JMS_CACHE_INVALIDATION_BRIDGE);
            msg.setObject(invalidationMsg);
            getPublisher().publish(msg);
         }
         catch (JMSException ex)
         {
            log.debug("failed to publish seppuku event: ", ex);
         }
      }
      catch (Exception ex)
      {
         log.warn("failed to do cluster seppuku event: " , ex);
      }
}

 public  void setAcknowledgeMode(int ackMode) 
{
      if (ackMode  > 3 || ackMode <  1)
         throw new RuntimeException ("Value AcknowledgeMode must be between 1 and 3");
      switch (ackMode)
      {
         case 1: this.acknowledgeMode = TopicSession.AUTO_ACKNOWLEDGE; break;
         case 2: this.acknowledgeMode = TopicSession.CLIENT_ACKNOWLEDGE; break;
         case 3: this.acknowledgeMode = TopicSession.DUPS_OK_ACKNOWLEDGE; break;
      }
}

 public  void setConnectionFactoryName(String factoryName) 
{
      this.connectionFactoryName = factoryName;
}

 public  void setInvalidationManager(String objectName) 
{
      this.invalidationManagerName = objectName;
}

 public  void setPropagationMode(int propMode) 
{
      if (propMode  > 3 || propMode <  1)
         throw new RuntimeException ("Value PropagationMode must be between 1 and 3");
      this.propagationMode = propMode;
}

 public  void setProviderUrl(String providerUrl) 
{
      this.providerUrl = providerUrl;
}

 public  void setTopicName(String topicName) 
{
      this.topicName = topicName;
}

 public  void setTransacted(boolean isTransacted) 
{
      this.transacted = isTransacted;
}

 protected  void startService() throws Exception 
{
      log.info("Starting JMS cache invalidation bridge");
      // Deal with the InvalidationManager first..
      //
      this.invalMgr = (org.jboss.cache.invalidation.InvalidationManagerMBean)
         org.jboss.system.Registry.lookup (this.invalidationManagerName);
      this.invalidationSubscription = invalMgr.registerBridgeListener (this);
      // deal with JMS next
      //
      InitialContext iniCtx = getInitialContext ();
      Object tmp = iniCtx.lookup(this.connectionFactoryName);
      TopicConnectionFactory tcf = (TopicConnectionFactory) tmp;
      conn = tcf.createTopicConnection();
      topic = (Topic) iniCtx.lookup(this.topicName);
      session = conn.createTopicSession(this.transacted,
                                        this.acknowledgeMode);
      conn.start();
      // Are we publisher, subscriber, or both?
      //
      if (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_ONLY_BRIDGE_PROPAGATION)
      {
         this.subscriber = session.createSubscriber(topic);     
         this.subscriber.setMessageListener(this);
      }
      if (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
            this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
      {
         this.pub = session.createPublisher(topic);
         this.publishingAuthorized = true;
      }
}

 protected  void stopService() 
{
      log.info ("Stoping JMS cache invalidation bridge");
      try
      {
         if (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
               this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_ONLY_BRIDGE_PROPAGATION)
         {
            subscriber.close();
         }
         if (this.propagationMode == JMSCacheInvalidationBridgeMBean.IN_OUT_BRIDGE_PROPAGATION ||
               this.propagationMode == JMSCacheInvalidationBridgeMBean.OUT_ONLY_BRIDGE_PROPAGATION)
         {
            this.publishingAuthorized = false;
            pub.close();
         }
         conn.stop();
         session.close();
         conn.close();
      }
      catch (Exception ex)
      {
         log.warn("Failed to stop JMS resources associated with the JMS bridge: ", ex);
      }
}