SofaRegistry extends Registry { @Override public List<ProviderGroup> subscribe(ConsumerConfig config) { ProviderInfoListener providerInfoListener = config.getProviderInfoListener(); String appName = config.getAppName(); if (!registryConfig.isSubscribe()) { if (LOGGER.isInfoEnabled(appName)) { LOGGER.infoWithApp(appName, LogCodes.getLog(LogCodes.INFO_REGISTRY_IGNORE)); } return null; } if (!config.isSubscribe()) { return null; } String serviceName = SofaRegistryHelper.buildListDataId(config, config.getProtocol()); SofaRegistrySubscribeCallback callback; Subscriber listSubscriber = subscribers.get(serviceName); Configurator attrSubscriber; if (listSubscriber != null && providerInfoListener != null) { callback = (SofaRegistrySubscribeCallback) listSubscriber.getDataObserver(); callback.addProviderInfoListener(serviceName, config, providerInfoListener); callback.handleDataToListener(serviceName, config, providerInfoListener); } else { callback = new SofaRegistrySubscribeCallback(); callback.addProviderInfoListener(serviceName, config, providerInfoListener); SubscriberRegistration subscriberRegistration = new SubscriberRegistration(serviceName, callback); String groupId = config.getParameter(SofaRegistryConstants.SOFA_GROUP_KEY); groupId = groupId == null ? SofaRegistryHelper.SUBSCRIBER_LIST_GROUP_ID : groupId; addAttributes(subscriberRegistration, groupId); ConfiguratorRegistration configRegistration = new ConfiguratorRegistration(serviceName, callback); addAttributes(configRegistration, SofaRegistryHelper.SUBSCRIBER_CONFIG_GROUP_ID); listSubscriber = SofaRegistryClient.getRegistryClient(appName, registryConfig).register( subscriberRegistration); attrSubscriber = SofaRegistryClient.getRegistryClient(appName, registryConfig).register(configRegistration); subscribers.put(serviceName, listSubscriber); configurators.put(serviceName, attrSubscriber); } return null; } protected  SofaRegistry(RegistryConfig registryConfig); @Override void init(); @Override void destroy(); @Override boolean start(); @Override void register(ProviderConfig config); @Override void unRegister(ProviderConfig config); @Override void batchUnRegister(List<ProviderConfig> configs); @Override List<ProviderGroup> subscribe(ConsumerConfig config); @Override void unSubscribe(ConsumerConfig config); @Override void batchUnSubscribe(List<ConsumerConfig> configs);  }
@Test public void testSubTwice() throws Exception { int timeoutPerSub = 5000; consumer1 = new ConsumerConfig(); consumer1.setInterfaceId("com.alipay.xxx.TestService") .setUniqueId("unique123Id") .setApplication(new ApplicationConfig().setAppName("test-server")) .setProxy("javassist") .setSubscribe(true) .setSerialization("java") .setInvokeType("sync") .setTimeout(4444); CountDownLatch latch = new CountDownLatch(2); MockProviderInfoListener providerInfoListener = new MockProviderInfoListener(); providerInfoListener.setCountDownLatch(latch); consumer1.setProviderInfoListener(providerInfoListener); registry.subscribe(consumer1); latch.await(timeoutPerSub, TimeUnit.MILLISECONDS); Map<String, ProviderGroup> ps = providerInfoListener.getData(); Assert.assertTrue(ps.size() > 0); Assert.assertNotNull(ps.get(RpcConstants.ADDRESS_DEFAULT_GROUP)); Assert.assertEquals(0, ps.get(RpcConstants.ADDRESS_DEFAULT_GROUP).size()); consumer2 = new ConsumerConfig(); consumer2.setInterfaceId("com.alipay.xxx.TestService") .setUniqueId("unique123Id") .setApplication(new ApplicationConfig().setAppName("test-server")) .setProxy("javassist") .setSubscribe(true) .setSerialization("java") .setInvokeType("sync") .setTimeout(4444); CountDownLatch latch2 = new CountDownLatch(1); MockProviderInfoListener providerInfoListener2 = new MockProviderInfoListener(); providerInfoListener2.setCountDownLatch(latch2); consumer2.setProviderInfoListener(providerInfoListener2); registry.subscribe(consumer2); latch2.await(timeoutPerSub, TimeUnit.MILLISECONDS); Map<String, ProviderGroup> ps2 = providerInfoListener2.getData(); Assert.assertTrue(ps2.size() > 0); Assert.assertNotNull(ps2.get(RpcConstants.ADDRESS_DEFAULT_GROUP)); Assert.assertEquals(0, ps2.get(RpcConstants.ADDRESS_DEFAULT_GROUP).size()); Assert.assertEquals(1, registry.subscribers.size()); Assert.assertEquals(1, registry.configurators.size()); latch.await(timeoutPerSub * 2, TimeUnit.MILLISECONDS); Assert.assertTrue(ps2.size() > 0); Assert.assertNotNull(ps2.get(RpcConstants.ADDRESS_DEFAULT_GROUP)); Assert.assertEquals(0, ps2.get(RpcConstants.ADDRESS_DEFAULT_GROUP).size()); Assert.assertEquals(1, registry.subscribers.size()); Assert.assertEquals(1, registry.configurators.size()); }