import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
/**** help method *****/
private void notifyLoadBalance(GrpcURL subscribeUrl, List<GrpcURL> urls) {
  if (urls != null && !urls.isEmpty()) {
    List<EquivalentAddressGroup> servers = Lists.newArrayList();
    List<SocketAddress> addresses = Lists.newArrayList();
    Map<List<SocketAddress>, GrpcURL> addressUrlMapping = Maps.newHashMap();
    for (GrpcURL url : urls) {
      String host = url.getHost();
      int port = url.getPort();
      List<SocketAddress> hostAddressMapping;
      if (NetUtils.isIP(host)) {
        hostAddressMapping = IpResolved(servers, addresses, host, port);
      } else {
        hostAddressMapping = DnsResolved(servers, addresses, host, port);
      }
      addressUrlMapping.put(hostAddressMapping, url);
    }
    this.addresses.put(subscribeUrl, addresses);
    Attributes config = this.buildAttributes(subscribeUrl, addressUrlMapping);
    GrpcNameResolver.this.listener.onAddresses(servers, config);
  } else {
    GrpcNameResolver.this.listener
        .onError(Status.NOT_FOUND.withDescription("There is no service registy in consul "));
  }
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
private List<SocketAddress> DnsResolved(List<EquivalentAddressGroup> servers,
    List<SocketAddress> addresses, String host, int port) {
  List<SocketAddress> hostAddressMapping = Lists.newArrayList();
  try {
    InetAddress[] inetAddrs = InetAddress.getAllByName(host);
    for (int j = 0; j < inetAddrs.length; j++) {
      InetAddress inetAddr = inetAddrs[j];
      SocketAddress sock = new InetSocketAddress(inetAddr, port);
      hostAddressMapping.add(sock);
      addSocketAddress(servers, addresses, sock);
    }
    return hostAddressMapping;
  } catch (UnknownHostException e) {
    GrpcNameResolver.this.listener.onError(Status.UNAVAILABLE.withCause(e));
  }
  return hostAddressMapping;
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Override
public List<EquivalentAddressGroup> resolve(URI uri) {
  if (!supports(uri)) {
    // Wrap as etcd exception but set a proper cause
    throw EtcdExceptionFactory.newEtcdException(
        ErrorCode.INVALID_ARGUMENT,
        "Unsupported URI " + uri
    );
  }

  return this.cache.computeIfAbsent(
      uri,
      u -> Collections.singletonList(
          new EquivalentAddressGroup(new InetSocketAddress(uri.getHost(), uri.getPort()))
      )
  );
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
private void update(Endpoints endpoints) {
    List<EquivalentAddressGroup> servers = new ArrayList<>();
    endpoints.getSubsets().stream().forEach(subset -> {
      long matchingPorts = subset.getPorts().stream().filter(p -> {
        return p.getPort() == port;
      }).count();
      if (matchingPorts > 0) {
        subset.getAddresses().stream().map(address -> {
          return new EquivalentAddressGroup(new InetSocketAddress(address.getIp(), port));
        }).forEach(address -> {
          servers.add(address);
        });
      }
    });

    listener.onAddresses(servers, Attributes.EMPTY);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Override
@Synchronized
public void start(Listener listener) {
    Preconditions.checkState(this.resolverUpdater == null, "ControllerNameResolver has already been started");
    Preconditions.checkState(!shutdown, "ControllerNameResolver is shutdown, restart is not supported");
    this.resolverUpdater = listener;

    // If the servers comprise only of IP addresses then we need to update the controller list only once.
    if (this.scheduledExecutor == null) {
        // Use the bootstrapped server list as the final set of controllers.
        List<EquivalentAddressGroup> servers = this.bootstrapServers.stream()
                .map(address -> new EquivalentAddressGroup(
                        new InetSocketAddress(address.getHostString(), address.getPort())))
                .collect(Collectors.toList());
        log.info("Updating client with controllers: {}", servers);
        this.resolverUpdater.onAddresses(servers, Attributes.EMPTY);
        return;
    }

    // Schedule the first discovery immediately.
    this.scheduledFuture = this.scheduledExecutor.schedule(this::getControllers, 0L, TimeUnit.SECONDS);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
/**
 * Handle new addresses of the balancer and backends from the resolver, and create connection if
 * not yet connected.
 */
void handleAddresses(
    List<LbAddressGroup> newLbAddressGroups, List<EquivalentAddressGroup> newBackendServers) {
  if (newLbAddressGroups.isEmpty()) {
    propagateError(Status.UNAVAILABLE.withDescription(
            "NameResolver returned no LB address while asking for GRPCLB"));
    return;
  }
  LbAddressGroup newLbAddressGroup = flattenLbAddressGroups(newLbAddressGroups);
  startLbComm(newLbAddressGroup);
  // Avoid creating a new RPC just because the addresses were updated, as it can cause a
  // stampeding herd. The current RPC may be on a connection to an address not present in
  // newLbAddressGroups, but we're considering that "okay". If we detected the RPC is to an
  // outdated backend, we could choose to re-create the RPC.
  if (lbStream == null) {
    startLbRpc();
  }
  fallbackBackendList = newBackendServers;
  maybeStartFallbackTimer();
  if (usingFallbackBackends) {
    // Populate the new fallback backends to round-robin list.
    useFallbackBackends();
  }
  maybeUpdatePicker();
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void nameResolutionFailsThenRecoverToDelegate() {
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
  RoundRobinPicker picker = (RoundRobinPicker) pickerCaptor.getValue();
  assertThat(picker.dropList).isEmpty();
  assertThat(picker.pickList).containsExactly(new ErrorEntry(error));

  // Recover with a subsequent success
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false);

  Attributes resolutionAttrs = Attributes.newBuilder().set(RESOLUTION_ATTR, "yeah").build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(pickFirstBalancerFactory).newLoadBalancer(helper);
  verify(pickFirstBalancer).handleResolvedAddressGroups(eq(resolvedServers), eq(resolutionAttrs));
  verifyNoMoreInteractions(roundRobinBalancerFactory);
  verifyNoMoreInteractions(roundRobinBalancer);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void delegatingPickFirstThenNameResolutionFails() {
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false);

  Attributes resolutionAttrs = Attributes.newBuilder().set(RESOLUTION_ATTR, "yeah").build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(pickFirstBalancerFactory).newLoadBalancer(helper);
  verify(pickFirstBalancer).handleResolvedAddressGroups(eq(resolvedServers), eq(resolutionAttrs));

  // Then let name resolution fail.  The error will be passed directly to the delegate.
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(pickFirstBalancer).handleNameResolutionError(error);
  verify(helper, never())
      .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
  verifyNoMoreInteractions(roundRobinBalancerFactory);
  verifyNoMoreInteractions(roundRobinBalancer);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void delegatingRoundRobinThenNameResolutionFails() {
  List<EquivalentAddressGroup> resolvedServers = createResolvedServerAddresses(false, false);

  Attributes resolutionAttrs = Attributes.newBuilder()
      .set(RESOLUTION_ATTR, "yeah")
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.ROUND_ROBIN)
      .build();
  deliverResolvedAddresses(resolvedServers, resolutionAttrs);

  verify(roundRobinBalancerFactory).newLoadBalancer(helper);
  verify(roundRobinBalancer).handleResolvedAddressGroups(resolvedServers, resolutionAttrs);

  // Then let name resolution fail.  The error will be passed directly to the delegate.
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);
  verify(roundRobinBalancer).handleNameResolutionError(error);
  verify(helper, never())
      .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
  verifyNoMoreInteractions(pickFirstBalancerFactory);
  verifyNoMoreInteractions(pickFirstBalancer);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void grpclbUpdatedAddresses_avoidsReconnect() {
  List<EquivalentAddressGroup> grpclbResolutionList =
      createResolvedServerAddresses(true, false);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList.get(0)), eq(lbAuthority(0)));
  ManagedChannel oobChannel = fakeOobChannels.poll();
  assertEquals(1, lbRequestObservers.size());

  List<EquivalentAddressGroup> grpclbResolutionList2 =
      createResolvedServerAddresses(true, false, true);
  EquivalentAddressGroup combinedEag = new EquivalentAddressGroup(Arrays.asList(
      grpclbResolutionList2.get(0).getAddresses().get(0),
      grpclbResolutionList2.get(2).getAddresses().get(0)));
  deliverResolvedAddresses(grpclbResolutionList2, grpclbResolutionAttrs);
  verify(helper).updateOobChannelAddresses(eq(oobChannel), addrsEq(combinedEag));
  assertEquals(1, lbRequestObservers.size()); // No additional RPC
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void grpclbUpdatedAddresses_reconnectOnAuthorityChange() {
  List<EquivalentAddressGroup> grpclbResolutionList =
      createResolvedServerAddresses(true, false);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList.get(0)), eq(lbAuthority(0)));
  ManagedChannel oobChannel = fakeOobChannels.poll();
  assertEquals(1, lbRequestObservers.size());

  final String newAuthority = "some-new-authority";
  List<EquivalentAddressGroup> grpclbResolutionList2 =
      createResolvedServerAddresses(false);
  grpclbResolutionList2.add(new EquivalentAddressGroup(
      new FakeSocketAddress("somethingNew"), lbAttributes(newAuthority)));
  deliverResolvedAddresses(grpclbResolutionList2, grpclbResolutionAttrs);
  assertTrue(oobChannel.isTerminated());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList2.get(1)), eq(newAuthority));
  assertEquals(2, lbRequestObservers.size()); // An additional RPC
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Override
public NameResolver newNameResolver(URI notUsedUri, Attributes params) {
  return new NameResolver() {
    @Override
    public String getServiceAuthority() {
      return authority;
    }

    @Override
    public void start(final Listener listener) {
      listener.onAddresses(
          Collections.singletonList(new EquivalentAddressGroup(address)),
          Attributes.EMPTY);
    }

    @Override
    public void shutdown() {}
  };
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
InternalSubchannel(EquivalentAddressGroup addressGroup, String authority, String userAgent,
    BackoffPolicy.Provider backoffPolicyProvider,
    ClientTransportFactory transportFactory, ScheduledExecutorService scheduledExecutor,
    Supplier<Stopwatch> stopwatchSupplier, ChannelExecutor channelExecutor, Callback callback,
    ProxyDetector proxyDetector) {
  this.addressGroup = Preconditions.checkNotNull(addressGroup, "addressGroup");
  this.authority = authority;
  this.userAgent = userAgent;
  this.backoffPolicyProvider = backoffPolicyProvider;
  this.transportFactory = transportFactory;
  this.scheduledExecutor = scheduledExecutor;
  this.connectingTimer = stopwatchSupplier.get();
  this.channelExecutor = channelExecutor;
  this.callback = callback;
  this.proxyDetector = proxyDetector;
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Override
ResolutionResults resolve(String host) throws Exception {
  ResolutionResults jdkResults = jdkResovler.resolve(host);
  List<InetAddress> addresses = jdkResults.addresses;
  List<String> txtRecords = Collections.emptyList();
  List<EquivalentAddressGroup> balancerAddresses = Collections.emptyList();
  try {
    ResolutionResults jdniResults = jndiResovler.resolve(host);
    txtRecords = jdniResults.txtRecords;
    balancerAddresses = jdniResults.balancerAddresses;
  } catch (Exception e) {
    logger.log(Level.SEVERE, "Failed to resolve TXT results", e);
  }

  return new ResolutionResults(addresses, txtRecords, balancerAddresses);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void loadBalancerThrowsInHandleResolvedAddresses() {
  RuntimeException ex = new RuntimeException("simulated");
  // Delay the success of name resolution until allResolved() is called
  FakeNameResolverFactory nameResolverFactory = new FakeNameResolverFactory(false);
  createChannel(nameResolverFactory, NO_INTERCEPTOR);

  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
  doThrow(ex).when(mockLoadBalancer).handleResolvedAddressGroups(
      Matchers.<List<EquivalentAddressGroup>>anyObject(), any(Attributes.class));

  // NameResolver returns addresses.
  nameResolverFactory.allResolved();

  // The LoadBalancer will receive the error that it has thrown.
  verify(mockLoadBalancer).handleNameResolutionError(statusCaptor.capture());
  Status status = statusCaptor.getValue();
  assertSame(Status.Code.INTERNAL, status.getCode());
  assertSame(ex, status.getCause());
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void doNotResolveWhenProxyDetected() throws Exception {
  final String name = "foo.googleapis.com";
  final int port = 81;
  ProxyDetector alwaysDetectProxy = mock(ProxyDetector.class);
  ProxyParameters proxyParameters = new ProxyParameters(
      InetSocketAddress.createUnresolved("proxy.example.com", 1000),
      "username",
      "password");
  when(alwaysDetectProxy.proxyFor(any(SocketAddress.class)))
      .thenReturn(proxyParameters);
  DelegateResolver unusedResolver = mock(DelegateResolver.class);
  DnsNameResolver resolver = newResolver(name, port, unusedResolver, alwaysDetectProxy);
  resolver.start(mockListener);
  assertEquals(1, fakeExecutor.runDueTasks());
  verify(unusedResolver, never()).resolve(any(String.class));

  verify(mockListener).onAddresses(resultCaptor.capture(), any(Attributes.class));
  List<EquivalentAddressGroup> result = resultCaptor.getValue();
  assertThat(result).hasSize(1);
  EquivalentAddressGroup eag = result.get(0);
  assertThat(eag.getAddresses()).hasSize(1);
  SocketAddress socketAddress = eag.getAddresses().get(0);
  assertTrue(((InetSocketAddress) socketAddress).isUnresolved());
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void updateSubchannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  Subchannel subchannel = helper.createSubchannel(servers.get(0), Attributes.EMPTY);

  subchannel.requestConnection();
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateSubchannelAddresses(subchannel, new EquivalentAddressGroup(changedList));

  subchannel.requestConnection();
  assertNull(newTransports.poll());
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void updateOobChannelAddresses_existingAddressDoesNotConnect() {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata()); // Create LB
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  ManagedChannel oobChannel = helper.createOobChannel(servers.get(0), "localhost");

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  List<SocketAddress> changedList = new ArrayList<SocketAddress>(servers.get(0).getAddresses());
  changedList.add(new FakeSocketAddress("aDifferentServer"));
  helper.updateOobChannelAddresses(oobChannel, new EquivalentAddressGroup(changedList));

  oobChannel.newCall(method, CallOptions.DEFAULT).start(mockCallListener, new Metadata());
  assertNull(newTransports.poll());
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
@Test
public void nameResolutionErrorWithActiveChannels() throws Exception {
  final Subchannel readySubchannel = subchannels.values().iterator().next();
  loadBalancer.handleResolvedAddressGroups(servers, affinity);
  loadBalancer.handleSubchannelState(readySubchannel, ConnectivityStateInfo.forNonError(READY));
  loadBalancer.handleNameResolutionError(Status.NOT_FOUND.withDescription("nameResolutionError"));

  verify(mockHelper, times(3)).createSubchannel(any(EquivalentAddressGroup.class),
      any(Attributes.class));
  verify(mockHelper, times(3))
      .updateBalancingState(stateCaptor.capture(), pickerCaptor.capture());

  Iterator<ConnectivityState> stateIterator = stateCaptor.getAllValues().iterator();
  assertEquals(CONNECTING, stateIterator.next());
  assertEquals(READY, stateIterator.next());
  assertEquals(TRANSIENT_FAILURE, stateIterator.next());

  LoadBalancer.PickResult pickResult = pickerCaptor.getValue().pickSubchannel(mockArgs);
  assertEquals(readySubchannel, pickResult.getSubchannel());
  assertEquals(Status.OK.getCode(), pickResult.getStatus().getCode());

  LoadBalancer.PickResult pickResult2 = pickerCaptor.getValue().pickSubchannel(mockArgs);
  assertEquals(readySubchannel, pickResult2.getSubchannel());
  verifyNoMoreInteractions(mockHelper);
}
 

import io.grpc.EquivalentAddressGroup; //导入依赖的package包/类
/**
 * Converts list of {@link EquivalentAddressGroup} to {@link EquivalentAddressGroup} set and
 * remove all attributes.
 */
private static Set<EquivalentAddressGroup> stripAttrs(List<EquivalentAddressGroup> groupList) {
  Set<EquivalentAddressGroup> addrs = new HashSet<EquivalentAddressGroup>();
  for (EquivalentAddressGroup group : groupList) {
    addrs.add(new EquivalentAddressGroup(group.getAddresses()));
  }
  return addrs;
}