ZookeeperRegistry extends Registry { protected void subscribeOverride(final ConsumerConfig config, ConfigListener listener) { try { if (overrideObserver == null) { overrideObserver = new ZookeeperOverrideObserver(); } overrideObserver.addConfigListener(config, listener); final String overridePath = buildOverridePath(rootPath, config); final AbstractInterfaceConfig registerConfig = getRegisterConfig(config); PathChildrenCache pathChildrenCache = new PathChildrenCache(zkClient, overridePath, true); pathChildrenCache.getListenable().addListener(new PathChildrenCacheListener() { @Override public void childEvent(CuratorFramework client1, PathChildrenCacheEvent event) throws Exception { if (LOGGER.isDebugEnabled(config.getAppName())) { LOGGER.debug("Receive zookeeper event: " + "type=[" + event.getType() + "]"); } switch (event.getType()) { case CHILD_ADDED: overrideObserver.addConfig(config, overridePath, event.getData()); break; case CHILD_REMOVED: overrideObserver.removeConfig(config, overridePath, event.getData(), registerConfig); break; case CHILD_UPDATED: overrideObserver.updateConfig(config, overridePath, event.getData()); break; default: break; } } }); pathChildrenCache.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE); INTERFACE_OVERRIDE_CACHE.put(overridePath, pathChildrenCache); overrideObserver.updateConfigAll(config, overridePath, pathChildrenCache.getCurrentData()); } catch (Exception e) { throw new SofaRpcRuntimeException(LogCodes.getLog(LogCodes.ERROR_SUB_PROVIDER_OVERRIDE, EXT_NAME), e); } } protected  ZookeeperRegistry(RegistryConfig registryConfig); @Override synchronized void init(); @Override synchronized boolean start(); @Override void destroy(); @Override void destroy(DestroyHook hook); @Override void register(ProviderConfig config); @Override void unRegister(ProviderConfig config); @Override void batchUnRegister(List<ProviderConfig> configs); @Override List<ProviderGroup> subscribe(final ConsumerConfig config); @Override void unSubscribe(ConsumerConfig config); @Override void batchUnSubscribe(List<ConsumerConfig> configs); static final String EXT_NAME; final static String PARAM_PREFER_LOCAL_FILE; final static String PARAM_CREATE_EPHEMERAL; }
@Test public void testOverrideObserver() throws InterruptedException { ConsumerConfig<?> consumerConfig = new ConsumerConfig(); consumerConfig.setInterfaceId(TEST_SERVICE_NAME) .setUniqueId("unique123Id") .setApplication(new ApplicationConfig().setAppName("test-server")) .setProxy("javassist") .setSubscribe(true) .setSerialization("java") .setInvokeType("sync") .setTimeout(4444); CountDownLatch latch = new CountDownLatch(1); MockConfigListener configListener = new MockConfigListener(); configListener.setCountDownLatch(latch); registry.subscribeOverride(consumerConfig, configListener); Map<String, String> attributes = new ConcurrentHashMap<String, String>(); attributes.put(RpcConstants.CONFIG_KEY_TIMEOUT, "3333"); attributes.put(RpcConstants.CONFIG_KEY_APP_NAME, "test-server"); attributes.put(RpcConstants.CONFIG_KEY_SERIALIZATION, "java"); configListener.attrUpdated(attributes); Map<String, String> configData = configListener.getData(); Assert.assertEquals(3, configData.size()); consumerConfig.setInterfaceId(TEST_SERVICE_NAME) .setUniqueId("unique123Id") .setApplication(new ApplicationConfig().setAppName("test-server1")) .setProxy("javassist") .setSubscribe(true) .setSerialization("java") .setInvokeType("sync") .setTimeout(5555); configListener = new MockConfigListener(); configListener.setCountDownLatch(latch); registry.subscribeOverride(consumerConfig, configListener); attributes.put(RpcConstants.CONFIG_KEY_TIMEOUT, "4444"); attributes.put(RpcConstants.CONFIG_KEY_APP_NAME, "test-server2"); configListener.attrUpdated(attributes); configData = configListener.getData(); Assert.assertEquals(3, configData.size()); latch.await(2000, TimeUnit.MILLISECONDS); Assert.assertEquals(3, configData.size()); }