ConsulNamingService extends  FailbackNamingService implements NamingService { @Override public List<ServiceInstance> lookup(SubscribeInfo subscribeInfo) { try { Response<List<HealthService>> consulServices = lookupHealthService( subscribeInfo.getServiceId(), -1); List<ServiceInstance> instances = convert(consulServices); log.info("lookup {} instances from consul", instances.size()); return instances; } catch (Exception ex) { log.warn("lookup endpoint list failed from {}, msg={}", url, ex.getMessage()); if (!subscribeInfo.isIgnoreFailOfNamingService()) { throw new RpcException("lookup endpoint list failed from consul failed", ex); } else { return new ArrayList<ServiceInstance>(); } } } ConsulNamingService(BrpcURL url); @Override void destroy(); @Override List<ServiceInstance> lookup(SubscribeInfo subscribeInfo); @Override void doSubscribe(SubscribeInfo subscribeInfo, NotifyListener listener); @Override void doUnsubscribe(SubscribeInfo subscribeInfo); @Override void doRegister(RegisterInfo registerInfo); @Override void doUnregister(RegisterInfo registerInfo); String generateInstanceId(RegisterInfo registerInfo); Response<List<HealthService>> lookupHealthService(String serviceName, long lastConsulIndex); List<ServiceInstance> convert(Response<List<HealthService>> consulServices);  }
@Test public void testLookup() throws InterruptedException { SubscribeInfo subscribeInfo = createSubscribeInfo(true); List<ServiceInstance> instances = consulNamingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 0); RegisterInfo registerInfo = createRegisterInfo("127.0.0.1", 8012); consulNamingService.register(registerInfo); Thread.sleep(3 * 1000); instances = consulNamingService.lookup(subscribeInfo); Assert.assertTrue(instances.size() == 1); Assert.assertTrue(instances.get(0).getIp().equals("127.0.0.1")); Assert.assertTrue(instances.get(0).getPort() == 8012); consulNamingService.unregister(registerInfo); }
@Test public void testRegisterAndSubscribe() throws InterruptedException { RegisterInfo registerInfo = createRegisterInfo("127.0.0.1", 8015); RegisterInfo anotherRegisterInfo = createRegisterInfo("127.0.0.1", 8016); consulNamingService.register(registerInfo); consulNamingService.register(anotherRegisterInfo); Thread.sleep(3 * 1000); SubscribeInfo subscribeInfo = createSubscribeInfo(false); final List<ServiceInstance> adds = consulNamingService.lookup(subscribeInfo); Assert.assertTrue(adds.size() == 2); adds.clear(); final List<ServiceInstance> deletes = new ArrayList<ServiceInstance>(); consulNamingService.subscribe(subscribeInfo, new NotifyListener() { @Override public void notify(Collection<ServiceInstance> addList, Collection<ServiceInstance> deleteList) { adds.addAll(addList); deletes.addAll(deleteList); log.info("notify: {}, {}", addList, deleteList); } }); consulNamingService.unregister(registerInfo); Thread.sleep(3 * 1000); Assert.assertTrue(adds.size() == 0); Assert.assertTrue(deletes.size() == 1); Assert.assertTrue(deletes.get(0).equals(new ServiceInstance("127.0.0.1", 8015))); adds.clear(); deletes.clear(); consulNamingService.register(registerInfo); Thread.sleep(3 * 1000); Assert.assertTrue(adds.size() == 1); Assert.assertTrue(deletes.size() == 0); Assert.assertTrue(adds.get(0).equals(new ServiceInstance("127.0.0.1", 8015))); }