SyncManager extends AbstractSyncManager { protected void doUpdateConfiguration() throws FloodlightModuleException { try { ClusterConfig oldConfig = clusterConfig; clusterConfig = clusterConfigProvider.getConfig(); if (clusterConfig.equals(oldConfig)) return; logger.info("[{}] Updating sync configuration {}", clusterConfig.getNode().getNodeId(), clusterConfig); if (oldConfig.getNode() != null && !clusterConfig.getNode().equals(oldConfig.getNode())) { logger.info("[{}] Local node configuration changed; restarting sync" + "service", oldConfig.getNode().getNodeId()); shutdown(); startUp(null); } for (Node n : clusterConfig.getNodes()) { Node existing = oldConfig.getNode(n.getNodeId()); if (existing != null && !n.equals(existing)) { logger.info("[{}->{}] Configuration for node has changed", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } for (Node n : oldConfig.getNodes()) { Node nn = clusterConfig.getNode(n.getNodeId()); if (nn == null) { logger.info("[{}->{}] Disconnecting deconfigured node", getLocalNodeId(), n.getNodeId()); rpcService.disconnectNode(n.getNodeId()); } } } catch (Exception e) { throw new FloodlightModuleException("Could not update " + "configuration", e); } }  @Override void registerStore(String storeName, Scope scope); @Override void registerPersistentStore(String storeName, Scope scope); ClusterConfig getClusterConfig(); void cleanup(); void antientropy(Node node); void antientropy(); void writeSyncValue(String storeName, Scope scope,
			boolean persist,
			byte[] key, Iterable<Versioned<byte[]>> values); boolean handleSyncOffer(String storeName,
			byte[] key,
			Iterable<VectorClock> versions); IStorageEngine<ByteArray, byte[]> getRawStore(String storeName); IThreadPoolService getThreadPool(); void queueSyncTask(SynchronizingStorageEngine e,
			ByteArray key, Versioned<byte[]> value); @Override void addListener(String storeName, MappingStoreListener listener); void updateConfiguration(); Cursor getCursor(int cursorId); Cursor newCursor(String storeName); void closeCursor(Cursor cursor); @Override IStore<ByteArray,byte[]> getStore(String storeName); @Override short getLocalNodeId(); @Override void shutdown(); @Override void init(FloodlightModuleContext context); @Override void startUp(FloodlightModuleContext context); @Override Collection<Class<? extends IFloodlightService>> getModuleDependencies(); static IDebugCounter counterHints; static IDebugCounter counterSentValues; static IDebugCounter counterReceivedValues; static IDebugCounter counterPuts; static IDebugCounter counterGets; static IDebugCounter counterIterators; static IDebugCounter counterErrorRemote; static IDebugCounter counterErrorProcessing; }
@Test public void testAddNode() throws Exception { waitForFullMesh(2000); IStoreClient<String, String> client0 = syncManagers[0].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client1 = syncManagers[1].getStoreClient("global", String.class, String.class); client0.put("key", "value"); waitForValue(client1, "key", "value", 2000, "client1"); nodes.add(new Node("localhost", 40105, (short)5, (short)5)); SyncManager[] sms = Arrays.copyOf(syncManagers, syncManagers.length + 1); FloodlightModuleContext[] fmcs = Arrays.copyOf(moduleContexts, moduleContexts.length + 1); sms[syncManagers.length] = new SyncManager(); fmcs[moduleContexts.length] = new FloodlightModuleContext(); nodeString = mapper.writeValueAsString(nodes); setupSyncManager(fmcs[moduleContexts.length], sms[syncManagers.length], nodes.get(syncManagers.length)); syncManagers = sms; moduleContexts = fmcs; for(int i = 0; i < 4; i++) { moduleContexts[i].addConfigParam(syncManagers[i], "nodes", nodeString); syncManagers[i].doUpdateConfiguration(); } waitForFullMesh(2000); IStoreClient<String, String> client4 = syncManagers[4].getStoreClient("global", String.class, String.class); client4.put("newkey", "newvalue"); waitForValue(client4, "key", "value", 2000, "client4"); waitForValue(client0, "newkey", "newvalue", 2000, "client0"); }
@Test public void testChangeNode() throws Exception { waitForFullMesh(2000); IStoreClient<String, String> client0 = syncManagers[0].getStoreClient("global", String.class, String.class); IStoreClient<String, String> client2 = syncManagers[2].getStoreClient("global", String.class, String.class); client0.put("key", "value"); waitForValue(client2, "key", "value", 2000, "client2"); nodes.set(2, new Node("localhost", 50103, (short)3, (short)1)); nodeString = mapper.writeValueAsString(nodes); for(int i = 0; i < syncManagers.length; i++) { moduleContexts[i].addConfigParam(syncManagers[i], "nodes", nodeString); syncManagers[i].doUpdateConfiguration(); } waitForFullMesh(2000); waitForValue(client2, "key", "value", 2000, "client2"); client2 = syncManagers[2].getStoreClient("global", String.class, String.class); client0.put("key", "newvalue"); waitForValue(client2, "key", "newvalue", 2000, "client2"); }