DefaultCurrentMetaManager extends AbstractLifecycleObservable implements CurrentMetaManager, Observer { @Override public void setPeerMaster(String dcId, String clusterId, String shardId, long gid, String ip, int port) { if (dcMetaCache.getCurrentDc().equalsIgnoreCase(dcId)) { throw new IllegalArgumentException(String.format("peer master must from other dc %s %s %d %s:%d", clusterId, shardId, gid, ip, port)); } RedisMeta peerMaster = new RedisMeta().setIp(ip).setPort(port).setGid(gid); currentMeta.setPeerMaster(dcId, clusterId, shardId, peerMaster); notifyPeerMasterChange(dcId, clusterId, shardId); } DefaultCurrentMetaManager(); @Override synchronized void addObserver(Observer observer); @Override Set<String> allClusters(); @Override void deleteSlot(int slotId); @Override void addSlot(int slotId); @Override void exportSlot(int slotId); @Override void importSlot(int slotId); @Override void update(Object args, Observable observable); @Override boolean hasCluster(String clusterId); @Override boolean hasShard(String clusterId, String shardId); @Override RedisMeta getRedisMaster(String clusterId, String shardId); @Override ClusterMeta getClusterMeta(String clusterId); @Override RouteMeta randomRoute(String clusterId); @Override Pair<String, Integer> getKeeperMaster(String clusterId, String shardId); @Override List<KeeperMeta> getSurviveKeepers(String clusterId, String shardId); @Override KeeperMeta getKeeperActive(String clusterId, String shardId); @Override String getCurrentMetaDesc(); @Override boolean updateKeeperActive(String clusterId, String shardId, KeeperMeta activeKeeper); @Override void addResource(String clusterId, String shardId, Releasable releasable); @Override void setSurviveKeepers(String clusterId, String shardId, List<KeeperMeta> surviceKeepers, KeeperMeta activeKeeper); @Override void setKeeperMaster(String clusterId, String shardId, String ip, int port); @Override void setKeeperMaster(String clusterId, String shardId, String addr); @Override boolean watchIfNotWatched(String clusterId, String shardId); @Override void setCurrentCRDTMaster(String clusterId, String shardId, long gid, String ip, int port); @Override RedisMeta getCurrentCRDTMaster(String clusterId, String shardId); @Override RedisMeta getCurrentMaster(String clusterId, String shardId); @Override void setPeerMaster(String dcId, String clusterId, String shardId, long gid, String ip, int port); @Override RedisMeta getPeerMaster(String dcId, String clusterId, String shardId); @Override Set<String> getUpstreamPeerDcs(String clusterId, String shardId); List<RedisMeta> getAllPeerMasters(String clusterId, String shardId); @Override void removePeerMaster(String dcId, String clusterId, String shardId); void setSlotManager(SlotManager slotManager); void setDcMetaCache(DcMetaCache dcMetaCache);  }
@Test public void testSetPeerMaster() { currentMetaServerMetaManager.addMetaServerStateChangeHandler(handler); currentMetaServerMetaManager.setCurrentMeta(currentMeta); doAnswer(invocation -> { String paramClusterId = invocation.getArgumentAt(0, String.class); String paramShardId = invocation.getArgumentAt(1, String.class); RedisMeta paramRedis = invocation.getArgumentAt(2, RedisMeta.class); Assert.assertEquals(getClusterId(), paramClusterId); Assert.assertEquals(getShardId(), paramShardId); Assert.assertEquals(new RedisMeta().setGid(1L).setIp("127.0.0.1").setPort(6379), paramRedis); return null; }).when(currentMeta).setCurrentCRDTMaster(anyString(), anyString(), any()); doAnswer(invocation -> { String paramDcId = invocation.getArgumentAt(0, String.class); String paramClusterId = invocation.getArgumentAt(1, String.class); String paramShardId = invocation.getArgumentAt(2, String.class); RedisMeta paramRedis = invocation.getArgumentAt(3, RedisMeta.class); Assert.assertEquals(upstreamDc, paramDcId); Assert.assertEquals(getClusterId(), paramClusterId); Assert.assertEquals(getShardId(), paramShardId); Assert.assertEquals(new RedisMeta().setGid(2L).setIp("127.0.0.2").setPort(6379), paramRedis); return null; }).when(currentMeta).setPeerMaster(anyString(), anyString(), anyString(), any()); currentMetaServerMetaManager.setCurrentCRDTMaster(getClusterId(), getShardId(), 1, "127.0.0.1", 6379); verify(currentMeta, times(1)).setCurrentCRDTMaster(anyString(), anyString(), any()); verify(handler, times(1)).currentMasterChanged(getClusterId(), getShardId()); currentMetaServerMetaManager.setPeerMaster(upstreamDc, getClusterId(), getShardId(), 2, "127.0.0.2", 6379); verify(currentMeta, times(1)).setPeerMaster(anyString(), anyString(), anyString(), any()); verify(handler, times(1)).peerMasterChanged(upstreamDc, getClusterId(), getShardId()); }
@Test(expected = IllegalArgumentException.class) public void testSetCurrentDcPeerMaster() { currentMetaServerMetaManager.setPeerMaster(getDc(), getClusterId(), getShardId(), 1, "127.0.0.1", 6379); }