public JGCacheInvalidationBridge() 
{
   // Static --------------------------------------------------------
   // Constructors --------------------------------------------------
}

 protected  void _do_rpc_batchInvalidate(BatchInvalidation[] invalidations,
    boolean asynch) 
{
      Object[] params = new Object[] {invalidations};
      try
      {         
         if (asynch)
            partition.callAsynchMethodOnCluster (this.RPC_HANDLER_NAME,
                                                      "_rpc_batchInvalidate", params, rpc_batch_invalidate_types, true);
         else
            partition.callMethodOnCluster (this.RPC_HANDLER_NAME,
                                                "_rpc_batchInvalidate", params, rpc_batch_invalidate_types, true);
      }
      catch (Exception e)
      {
         log.debug ("Distributed invalidation (3) has failed (Bridge: " + this.bridgeName + ")");
      }
}

 protected  void _do_rpc_invalidate(String invalidationGroupName,
    Serializable key,
    boolean asynch) 
{
      Object[] params = new Object[] {invalidationGroupName, key};
      try
      {         
         if (asynch)
            partition.callAsynchMethodOnCluster (this.RPC_HANDLER_NAME,
                                                      "_rpc_invalidate",
                                                      params, rpc_invalidate_types, true);
         else
            partition.callMethodOnCluster (this.RPC_HANDLER_NAME,
                                                "_rpc_invalidate",
                                                params, rpc_invalidate_types, true);
      }
      catch (Exception e)
      {
         log.debug ("Distributed invalidation (1) has failed for group " + 
                    invalidationGroupName + " (Bridge: " + this.bridgeName + ")");
      }
}

 protected  void _do_rpc_invalidate_all(String invalidationGroupName,
    boolean asynch) 
{
      Object[] params = new Object[] {invalidationGroupName};
      try
      {
         if (asynch)
            partition.callAsynchMethodOnCluster (this.RPC_HANDLER_NAME,
                                                      "_rpc_invalidate_all", params, rpc_invalidate_all_types, true);
         else
            partition.callMethodOnCluster (this.RPC_HANDLER_NAME,
                                                "_rpc_invalidate_all", params, rpc_invalidate_all_types, true);
      }
      catch (Exception e)
      {
         log.debug ("Distributed invalidation (2) has failed for group " +
                    invalidationGroupName + " (Bridge: " + this.bridgeName + ")");
      }
}

 protected  void _do_rpc_invalidates(String invalidationGroupName,
    Serializable[] keys,
    boolean asynch) 
{
      Object[] params = new Object[] {invalidationGroupName, keys};
      try
      {         
         if (asynch)
            partition.callAsynchMethodOnCluster (this.RPC_HANDLER_NAME,
                                                      "_rpc_invalidates", params, rpc_invalidates_types, true);
         else
            partition.callMethodOnCluster (this.RPC_HANDLER_NAME,
                                                "_rpc_invalidates", params, rpc_invalidates_types, true);
      }
      catch (Exception e)
      {
         log.debug ("Distributed invalidation (2) has failed for group " + 
                    invalidationGroupName + " (Bridge: " + this.bridgeName + ")");
      }
}

 public  void _rpc_batchInvalidate(BatchInvalidation[] invalidations) 
{
      if (log.isTraceEnabled () && invalidations != null)
         log.trace ("Received remote batch invalidation for this number of groups: " + invalidations.length);
      this.invalidationSubscription.batchInvalidate (invalidations);
}

 public  void _rpc_invalidate(String invalidationGroupName,
    Serializable key) 
{
      if (log.isTraceEnabled ())
         log.trace ("Received remote invalidation for group: " + invalidationGroupName);
      this.invalidationSubscription.invalidate (invalidationGroupName, key);
}

 public  void _rpc_invalidate_all(String invalidationGroupName) 
{
      if (log.isTraceEnabled ())
         log.trace ("Received remote invalidate_all for group: " + invalidationGroupName);
      this.invalidationSubscription.invalidateAll (invalidationGroupName);
}

 public  void _rpc_invalidates(String invalidationGroupName,
    Serializable[] keys) 
{
      if (log.isTraceEnabled ())
         log.trace ("Received remote invalidations for group: " + invalidationGroupName);
      this.invalidationSubscription.invalidate (invalidationGroupName, keys);
}

 public  void batchInvalidate(BatchInvalidation[] invalidations,
    boolean asynchronous) 
{
      if (invalidations == null) return;
      // we need to sort which group other nodes accept or refuse and propagate through the net
      //      
      ArrayList acceptedGroups = new ArrayList();
      for (int i=0; i< invalidations.length; i++)
      {
         BatchInvalidation currBI = invalidations[i];
         if (groupExistsRemotely (currBI.getInvalidationGroupName ()))
            acceptedGroups.add (currBI);
      }
      if (acceptedGroups.size ()  > 0)
      {
         BatchInvalidation[] result = new BatchInvalidation[acceptedGroups.size ()];
         result = (BatchInvalidation[])acceptedGroups.toArray (result);
         if (log.isTraceEnabled ())
            log.trace ("Transmitting batch invalidation: " + result);
         this._do_rpc_batchInvalidate (result, asynchronous);      
      }
}

 public String getBridgeName() 
{
      return this.bridgeName;
}

 public HAPartition getHAPartition() 
{
      return partition;
}

 public InvalidationManagerMBean getInvalidationManager() 
{
      return this.invalMgr;
}

 public String getPartitionName() 
{
      return this.partition.getPartitionName();
}

 protected boolean groupExistsRemotely(String groupName) 
{
      return this.bridgedGroups.contains (groupName);
}

 public  void groupIsDropped(String groupInvalidationName) 
{
      try
      {
         this.publishLocalInvalidationGroups ();
         //this.updatedBridgedInvalidationGroupsInfo ();
      }
      catch (Exception e)
      {
         log.info ("Problem while un-registering a new invalidation group over the cluster", e);
      }
}

 public  void invalidate(String invalidationGroupName,
    Serializable[] keys,
    boolean asynchronous) 
{
      // if the group exists on another node, we simply propagate to other nodes
      //
      if (log.isTraceEnabled ())
         log.trace ("Transmitting invalidations for group: " + invalidationGroupName);
      if (groupExistsRemotely (invalidationGroupName))
         _do_rpc_invalidates (invalidationGroupName, keys, asynchronous);
}

 public  void invalidate(String invalidationGroupName,
    Serializable key,
    boolean asynchronous) 
{
      // if the group exists on another node, we simply propagate to other nodes
      //
      if (log.isTraceEnabled ())
         log.trace ("Transmitting invalidation for group: " + invalidationGroupName);
      if (groupExistsRemotely (invalidationGroupName))
         _do_rpc_invalidate (invalidationGroupName, key, asynchronous);
}

 public  void invalidateAll(String groupName,
    boolean async) 
{
      if (log.isTraceEnabled ())
         log.trace ("Transmitting for all entries for invalidation for group: " + groupName);
      if (groupExistsRemotely (groupName))
         _do_rpc_invalidate_all (groupName, async);
}

 public  void keyHasBeenRemoved(String category,
    Serializable key,
    Serializable previousContent,
    boolean locallyModified) 
{
       this.updatedBridgedInvalidationGroupsInfo ();
}

 public  void newGroupCreated(String groupInvalidationName) 
{
      try
      {
         this.publishLocalInvalidationGroups ();
         //this.updatedBridgedInvalidationGroupsInfo ();
      }
      catch (Exception e)
      {
         log.info ("Problem while registering a new invalidation group over the cluster", e);
      }
}

 protected synchronized  void publishLocalInvalidationGroups() throws Exception 
{
      this.localGroups = invalMgr.getInvalidationGroups ();      
      log.debug ("Publishing locally available invalidation groups: " + this.localGroups);
      ArrayList content = new ArrayList (this.localGroups);      
      ArrayList result = new ArrayList (content.size ());
      for (int i = 0; i <  content.size(); i++)
      {
         String aGroup = ((InvalidationGroup)content.get(i)).getGroupName ();
         result.add (aGroup);
      }
      if (result.size ()  > 0)       
      {
         NodeInfo info = new NodeInfo (result, this.nodeName);
         this.ds.set (this.RPC_HANDLER_NAME, this.nodeName, info, true);
      }
      else
         this.ds.remove (this.RPC_HANDLER_NAME, this.nodeName, true);
}

 public synchronized  void replicantsChanged(String key,
    List newReplicants,
    int newReplicantsViewId,
    boolean merge) 
{
      if (key.equals (this.RPC_HANDLER_NAME) && this.drm.isMasterReplica (this.RPC_HANDLER_NAME))
      {
         log.debug ("The list of replicants for the JG bridge has changed, computing and updating local info...");
         // we remove any entry from the DS whose node is dead
         //
         java.util.Collection coll = this.ds.getAllKeys (this.RPC_HANDLER_NAME);                  
         if (coll == null) 
         {
            log.debug ("... No bridge info was associated with this node");
            return;
         }
         // to avoid ConcurrentModificationException, we copy the list of keys in a new structure
         //
         ArrayList collCopy = new java.util.ArrayList (coll);
         java.util.List newReplicantsNodeNames = this.drm.lookupReplicantsNodeNames (this.RPC_HANDLER_NAME);
         for (int i = 0; i <  collCopy.size(); i++)
         {
            String nodeEntry = (String)collCopy.get(i);
            if (!newReplicantsNodeNames.contains (nodeEntry))
            {
               // the list of bridged topic contains a dead member: we remove it
               //
               try
               {
                  log.debug ("removing bridge information associated with this node from the DS");
                  this.ds.remove (this.RPC_HANDLER_NAME, nodeEntry, true);
               }
               catch (Exception e)
               {
                  log.info ("Unable to remove a node entry from the distributed cache", e);
               }
            }
         }
      }
}

 public  void setBridgeName(String name) 
{
      this.bridgeName = name;
}

 public  void setHAPartition(HAPartition clusterPartition) 
{
      this.partition = clusterPartition;
}

 public  void setInvalidationManager(InvalidationManagerMBean manager) 
{
      this.invalMgr = manager;
}

 public  void startService() throws Exception 
{
      if (partition == null)
      {
         throw new IllegalStateException("HAPartition property must be set before starting InvalidationBridge service");
      }
      RPC_HANDLER_NAME = "DCacheBridge-" + this.bridgeName;
      this.ds = partition.getDistributedStateService ();
      this.drm = partition.getDistributedReplicantManager ();
      this.nodeName = partition.getNodeName();
      this.drm.add (this.RPC_HANDLER_NAME, "");
      this.drm.registerListener (this.RPC_HANDLER_NAME, this);      
      this.ds.registerDSListenerEx (RPC_HANDLER_NAME, this);
      partition.registerRPCHandler(RPC_HANDLER_NAME, this);   
      // we now publish the list of caches we have access to
      if( invalMgr == null )
         throw new IllegalStateException("Failed to find an InvalidationManagerMBean, ensure one is injected");
      publishLocalInvalidationGroups ();      
      this.updatedBridgedInvalidationGroupsInfo ();
      this.invalidationSubscription = invalMgr.registerBridgeListener (this);
}

 public  void stopService() 
{
      try
      {
         partition.unregisterRPCHandler (this.RPC_HANDLER_NAME, this);
         this.ds.unregisterDSListenerEx (this.RPC_HANDLER_NAME, this);
         this.drm.unregisterListener (this.RPC_HANDLER_NAME, this);
         this.drm.remove (this.RPC_HANDLER_NAME);
         this.invalidationSubscription.unregister ();
         this.ds.remove (this.RPC_HANDLER_NAME, this.nodeName, true);
//         this.invalMgr = null;
//         partition = null;
         this.drm = null;
         this.ds = null;
         this.invalidationSubscription = null;
         this.RPC_HANDLER_NAME = null;
         this.nodeName = null;
         this.localGroups = null;
         this.bridgedGroups = new Vector ();                  
      }
      catch (Exception e)
      {
         log.info ("Problem while shuting down invalidation cache bridge", e);
      }
}

 protected  void updatedBridgedInvalidationGroupsInfo() 
{
      Collection bridgedByNode = this.ds.getAllValues (this.RPC_HANDLER_NAME);
      log.debug ("Updating list of invalidation groups that are bridged...");
      if (bridgedByNode != null)
      {
         // Make a copy
         //      
         ArrayList copy = new ArrayList (bridgedByNode);
         Vector result  = new Vector ();
         for (int i = 0; i <  copy.size(); i++)
         {
            NodeInfo infoForNode = (NodeInfo)copy.get(i);
            log.trace ("InfoForNode: " + infoForNode);
            if (infoForNode != null && !infoForNode.groupName.equals (this.nodeName))
            {
               ArrayList groupsForNode = infoForNode.groups;
               log.trace ("Groups for node: " + groupsForNode);
               for (int j = 0; j <  groupsForNode.size(); j++)
               {
                  String aGroup = (String)groupsForNode.get(j);
                  if (!result.contains (aGroup))
                  {
                     log.trace ("Adding: " + aGroup);
                     result.add (aGroup);                  
                  }
               }
            }            
         }
         // atomic assignation of the result
         //
         this.bridgedGroups = result;
         log.debug ("... computed list of bridged groups: " + result);
      }
      else
      {
         log.debug ("... nothing needs to be bridged.");            
      }
}

 public  void valueHasChanged(String category,
    Serializable key,
    Serializable value,
    boolean locallyModified) 
{
       this.updatedBridgedInvalidationGroupsInfo ();
}