package org.jboss.cache.invalidation.bridges;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Vector;
import java.util.Collection;
import org.jboss.cache.invalidation.BatchInvalidation;
import org.jboss.cache.invalidation.InvalidationManager;
import org.jboss.cache.invalidation.InvalidationGroup;
import org.jboss.cache.invalidation.InvalidationManagerMBean;
import org.jboss.cache.invalidation.BridgeInvalidationSubscription;
import org.jboss.cache.invalidation.InvalidationBridgeListener;
import org.jboss.ha.framework.interfaces.HAPartition;
import org.jboss.ha.framework.interfaces.DistributedState;
import org.jboss.ha.framework.interfaces.DistributedReplicantManager;
import org.jboss.system.server.ServerConfigUtil;
public class JGCacheInvalidationBridge
extends org.jboss.system.ServiceMBeanSupport
implements JGCacheInvalidationBridgeMBean,
DistributedState.DSListenerEx,
InvalidationBridgeListener,
DistributedReplicantManager.ReplicantListener
{
protected String partitionName = ServerConfigUtil.getDefaultPartitionName();
protected String invalidationManagerName = InvalidationManager.DEFAULT_JMX_SERVICE_NAME;
protected String bridgeName = "DefaultJGCacheIB";
protected HAPartition partition = null;
protected DistributedState ds = null;
protected DistributedReplicantManager drm = null;
protected String RPC_HANLE_NAME = null;
protected String nodeName = null;
protected InvalidationManagerMBean invalMgr = null;
protected BridgeInvalidationSubscription invalidationSubscription = null;
protected Collection localGroups = null;
protected Vector bridgedGroups = new Vector ();
protected final Class[] rpc_invalidate_types=new Class[]{String.class, Serializable.class};
protected final Class[] rpc_invalidates_types=new Class[]{String.class, Serializable[].class};
protected final Class[] rpc_invalidate_all_types=new Class[]{String.class};
protected final Class[] rpc_batch_invalidate_types=new Class[]{BatchInvalidation[].class};
public JGCacheInvalidationBridge ()
{
}
public String getInvalidationManager ()
{
return this.invalidationManagerName;
}
public String getPartitionName ()
{
return this.partitionName;
}
public void setInvalidationManager (String objectName)
{
this.invalidationManagerName = objectName;
}
public void setPartitionName (String partitionName)
{
this.partitionName = partitionName;
}
public String getBridgeName ()
{
return this.bridgeName;
}
public void setBridgeName (String name)
{
this.bridgeName = name;
}
public void replicantsChanged (String key, java.util.List newReplicants, int newReplicantsViewId)
{
if (key.equals (this.RPC_HANLE_NAME) && this.drm.isMasterReplica (this.RPC_HANLE_NAME))
{
log.debug ("The list of replicant for the JG bridge has changed, computing and updating local info...");
java.util.Collection coll = this.ds.getAllKeys (this.RPC_HANLE_NAME);
if (coll == null)
{
log.debug ("... No bridge info was associated to this node");
return;
}
ArrayList collCopy = new java.util.ArrayList (coll);
java.util.List newReplicantsNodeNames = this.drm.lookupReplicantsNodeNames (this.RPC_HANLE_NAME);
for (int i = 0; i < collCopy.size(); i++)
{
String nodeEntry = (String)collCopy.get(i);
if (!newReplicantsNodeNames.contains (nodeEntry))
{
try
{
log.debug ("removing bridge information associated to this node from the DS");
this.ds.remove (this.RPC_HANLE_NAME, nodeEntry, true);
}
catch (Exception e)
{
log.info ("Unable to remove a node entry from the distributed cache", e);
}
}
}
}
}
public void valueHasChanged (String category, Serializable key, Serializable value, boolean locallyModified)
{
this.updatedBridgedInvalidationGroupsInfo ();
}
public void keyHasBeenRemoved (String category, Serializable key, Serializable previousContent, boolean locallyModified)
{
this.updatedBridgedInvalidationGroupsInfo ();
}
public void batchInvalidate (BatchInvalidation[] invalidations, boolean asynchronous)
{
if (invalidations == null) return;
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 void invalidate (String invalidationGroupName, Serializable[] keys, boolean asynchronous)
{
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 (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 newGroupCreated (String groupInvalidationName)
{
try
{
this.publishLocalInvalidationGroups ();
}
catch (Exception e)
{
log.info ("Problem while registering a new invalidation group over the cluster", e);
}
}
public void groupIsDropped (String groupInvalidationName)
{
try
{
this.publishLocalInvalidationGroups ();
}
catch (Exception e)
{
log.info ("Problem while un-registering a new invalidation group over the cluster", e);
}
}
public void startService () throws Exception
{
RPC_HANLE_NAME = "DCacheBridge-" + this.bridgeName;
javax.naming.Context ctx = new javax.naming.InitialContext ();
this.partition = (org.jboss.ha.framework.interfaces.HAPartition)ctx.lookup("/HAPartition/" + this.partitionName);
this.ds = this.partition.getDistributedStateService ();
this.drm = this.partition.getDistributedReplicantManager ();
this.nodeName = this.partition.getNodeName ();
this.drm.add (this.RPC_HANLE_NAME, "");
this.drm.registerListener (this.RPC_HANLE_NAME, this);
this.ds.registerDSListenerEx (RPC_HANLE_NAME, this);
this.partition.registerRPCHandler(RPC_HANLE_NAME, this);
this.invalMgr = (org.jboss.cache.invalidation.InvalidationManagerMBean)
org.jboss.system.Registry.lookup (this.invalidationManagerName);
publishLocalInvalidationGroups ();
this.updatedBridgedInvalidationGroupsInfo ();
this.invalidationSubscription = invalMgr.registerBridgeListener (this);
}
public void stopService ()
{
try
{
this.partition.unregisterRPCHandler (this.RPC_HANLE_NAME, this);
this.ds.unregisterDSListenerEx (this.RPC_HANLE_NAME, this);
this.drm.unregisterListener (this.RPC_HANLE_NAME, this);
this.drm.remove (this.RPC_HANLE_NAME);
this.invalidationSubscription.unregister ();
this.ds.remove (this.RPC_HANLE_NAME, this.nodeName, true);
this.invalMgr = null;
this.partition = null;
this.drm = null;
this.ds = null;
this.invalidationSubscription = null;
this.RPC_HANLE_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);
}
}
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_invalidates (String invalidationGroupName, Serializable[] keys)
{
if (log.isTraceEnabled ())
log.trace ("Received remote invalidations for group: " + invalidationGroupName);
this.invalidationSubscription.invalidate (invalidationGroupName, keys);
}
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_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);
}
protected void _do_rpc_invalidate (String invalidationGroupName, Serializable key, boolean asynch)
{
Object[] params = new Object[] {invalidationGroupName, key};
try
{
if (asynch)
this.partition.callAsynchMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_invalidate",
params, rpc_invalidate_types, true);
else
this.partition.callMethodOnCluster (this.RPC_HANLE_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_invalidates (String invalidationGroupName, Serializable[] keys, boolean asynch)
{
Object[] params = new Object[] {invalidationGroupName, keys};
try
{
if (asynch)
this.partition.callAsynchMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_invalidates", params, rpc_invalidates_types, true);
else
this.partition.callMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_invalidates", params, rpc_invalidates_types, true);
}
catch (Exception e)
{
log.debug ("Distributed invalidation (2) 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)
this.partition.callAsynchMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_invalidate_all", params, rpc_invalidate_all_types, true);
else
this.partition.callMethodOnCluster (this.RPC_HANLE_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_batchInvalidate (BatchInvalidation[] invalidations, boolean asynch)
{
Object[] params = new Object[] {invalidations};
try
{
if (asynch)
this.partition.callAsynchMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_batchInvalidate", params, rpc_batch_invalidate_types, true);
else
this.partition.callMethodOnCluster (this.RPC_HANLE_NAME,
"_rpc_batchInvalidate", params, rpc_batch_invalidate_types, true);
}
catch (Exception e)
{
log.debug ("Distributed invalidation (3) has failed (Bridge: " + this.bridgeName + ")");
}
}
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_HANLE_NAME, this.nodeName, info, true);
}
else
this.ds.remove (this.RPC_HANLE_NAME, this.nodeName, true);
}
protected void updatedBridgedInvalidationGroupsInfo ()
{
Collection bridgedByNode = this.ds.getAllValues (this.RPC_HANLE_NAME);
log.debug ("Updating list of invalidation groups that are bridged...");
if (bridgedByNode != null)
{
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);
}
}
}
}
this.bridgedGroups = result;
log.debug ("... computed list of bridged groups: " + result);
}
else
{
log.debug ("... nothing needs to be bridged.");
}
}
protected boolean groupExistsRemotely (String groupName)
{
return this.bridgedGroups.contains (groupName);
}
}
class NodeInfo implements java.io.Serializable
{
static final long serialVersionUID = -3215712955134929006L;
public ArrayList groups = null;
public String groupName = null;
public NodeInfo (){}
public NodeInfo (ArrayList groups, String groupName)
{
this.groups = groups;
this.groupName = groupName;
}
}