Ayaka Koshibe

notification for mastership changes

Change-Id: I191ccd42ff9f8a41e87cfcda07531e4fbdd923c8
package org.onlab.onos.store.cluster.impl;
import static com.google.common.cache.CacheBuilder.newBuilder;
import static org.onlab.onos.cluster.MastershipEvent.Type.MASTER_CHANGED;
import java.util.Map;
......@@ -20,12 +19,8 @@ import org.onlab.onos.cluster.MastershipTerm;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.net.DeviceId;
import org.onlab.onos.net.MastershipRole;
import org.onlab.onos.store.common.AbsentInvalidatingLoadingCache;
import org.onlab.onos.store.common.AbstractHazelcastStore;
import org.onlab.onos.store.common.OptionalCacheLoader;
import com.google.common.base.Optional;
import com.google.common.cache.LoadingCache;
import com.google.common.collect.ImmutableSet;
import com.hazelcast.core.ILock;
import com.hazelcast.core.IMap;
......@@ -53,9 +48,6 @@ implements MastershipStore {
//collection of nodes. values are ignored, as it's used as a makeshift 'set'
protected IMap<byte[], Byte> backups;
//TODO - remove
//private LoadingCache<DeviceId, Optional<NodeId>> masters;
@Reference(cardinality = ReferenceCardinality.MANDATORY_UNARY)
protected ClusterService clusterService;
......@@ -68,11 +60,7 @@ implements MastershipStore {
rawTerms = theInstance.getMap("terms");
backups = theInstance.getMap("backups");
//TODO: hook up maps to event notification
//OptionalCacheLoader<DeviceId, NodeId> nodeLoader
//= new OptionalCacheLoader<>(kryoSerializationService, rawMasters);
//masters = new AbsentInvalidatingLoadingCache<>(newBuilder().build(nodeLoader));
//rawMasters.addEntryListener(new RemoteMasterShipEventHandler(masters), true);
rawMasters.addEntryListener(new RemoteMasterShipEventHandler(), true);
log.info("Started");
}
......@@ -253,27 +241,26 @@ implements MastershipStore {
//adds node to pool(s) of backup
private void backup(NodeId nodeId, DeviceId deviceId) {
//TODO might be useful to isolate out
//TODO might be useful to isolate out this function and reelect() if we
//get more backup/election schemes
}
private class RemoteMasterShipEventHandler extends RemoteCacheEventHandler<DeviceId, NodeId> {
public RemoteMasterShipEventHandler(LoadingCache<DeviceId, Optional<NodeId>> cache) {
super(cache);
}
private class RemoteMasterShipEventHandler extends RemoteEventHandler<DeviceId, NodeId> {
@Override
protected void onAdd(DeviceId deviceId, NodeId nodeId) {
//only addition indicates a change in mastership
notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
}
@Override
protected void onRemove(DeviceId deviceId, NodeId nodeId) {
notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
//notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
}
@Override
protected void onUpdate(DeviceId deviceId, NodeId oldNodeId, NodeId nodeId) {
notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
//notifyDelegate(new MastershipEvent(MASTER_CHANGED, deviceId, nodeId));
}
}
......
......@@ -6,18 +6,23 @@ import static org.junit.Assert.assertTrue;
import static org.onlab.onos.net.MastershipRole.*;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
import org.onlab.onos.cluster.ClusterEventListener;
import org.onlab.onos.cluster.ClusterService;
import org.onlab.onos.cluster.ControllerNode;
import org.onlab.onos.cluster.ControllerNode.State;
import org.onlab.onos.cluster.DefaultControllerNode;
import org.onlab.onos.cluster.MastershipEvent;
import org.onlab.onos.cluster.MastershipEvent.Type;
import org.onlab.onos.cluster.MastershipStoreDelegate;
import org.onlab.onos.cluster.MastershipTerm;
import org.onlab.onos.cluster.NodeId;
import org.onlab.onos.net.DeviceId;
......@@ -40,7 +45,6 @@ public class DistributedMastershipStoreTest {
private static final DeviceId DID1 = DeviceId.deviceId("of:01");
private static final DeviceId DID2 = DeviceId.deviceId("of:02");
private static final DeviceId DID3 = DeviceId.deviceId("of:03");
private static final DeviceId DID4 = DeviceId.deviceId("of:04");
private static final IpPrefix IP = IpPrefix.valueOf("127.0.0.1");
......@@ -191,6 +195,28 @@ public class DistributedMastershipStoreTest {
assertEquals("wrong role for node:", NONE, dms.getRole(N1, DID2));
}
@Ignore("Ignore until Delegate spec. is clear.")
@Test
public void testEvents() throws InterruptedException {
//shamelessly copy other distributed store tests
final CountDownLatch addLatch = new CountDownLatch(1);
MastershipStoreDelegate checkAdd = new MastershipStoreDelegate() {
@Override
public void notify(MastershipEvent event) {
assertEquals("wrong event:", Type.MASTER_CHANGED, event.type());
assertEquals("wrong subject", DID1, event.subject());
assertEquals("wrong subject", N1, event.master());
addLatch.countDown();
}
};
dms.setDelegate(checkAdd);
dms.setMaster(N1, DID1);
//this will fail until we do something about single-instance-ness
assertTrue("Add event fired", addLatch.await(1, TimeUnit.SECONDS));
}
private class TestDistributedMastershipStore extends
DistributedMastershipStore {
public TestDistributedMastershipStore(StoreService storeService,
......