Java 类io.grpc.Attributes 实例源码

/**** 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 "));
  }
}

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);
}

@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);
}

@BeforeClass
public static void startServer() throws IOException {
    AfricasTalking.initialize(Fixtures.USERNAME, Fixtures.API_KEY);
    server = new Server(new Authenticator() {
        @Override
        public boolean authenticate(String client) {
            return client.compareToIgnoreCase(TEST_CLIENT_ID) == 0;
        }
    });
    server.addSipCredentials("test", "secret", "sip://at.dev");
    server.start(certFile, privateKeyFile, TEST_PORT);

    ManagedChannel ch = NettyChannelBuilder.forAddress("localhost", TEST_PORT)
        .sslContext(GrpcSslContexts.forClient().trustManager(certFile).build())
        .build();


    client = SdkServerServiceGrpc.newBlockingStub(ch)
        .withCallCredentials(new CallCredentials(){
            @Override
            public void applyRequestMetadata(MethodDescriptor<?, ?> method, Attributes attrs, Executor appExecutor,
                    final MetadataApplier applier) {
                        appExecutor.execute(new Runnable(){
                            @Override
                            public void run() {
                                try {
                                    Metadata headers = new Metadata();
                                    Metadata.Key<String> clientIdKey = Metadata.Key.of("X-Client-Id", Metadata.ASCII_STRING_MARSHALLER);
                                    headers.put(clientIdKey, TEST_CLIENT_ID);
                                    applier.apply(headers);
                                } catch(Throwable ex) {
                                    applier.fail(Status.UNAUTHENTICATED.withCause(ex));
                                }
                            }
                        });

            }
        });
}

@Nullable
@Override
public NameResolver newNameResolver(URI targetUri, Attributes params) {
  if (SCHEME.equals(targetUri.getScheme())) {
    String targetPath = Preconditions.checkNotNull(targetUri.getPath(), "targetPath");
    Preconditions.checkArgument(targetPath.startsWith("/"),
        "the path component (%s) of the target (%s) must start with '/'", targetPath, targetUri);

    String[] parts = targetPath.split("/");
    if (parts.length != 4) {
      throw new IllegalArgumentException("Must be formatted like kubernetes:///{namespace}/{service}/{port}");
    }

    try {
      int port = Integer.valueOf(parts[3]);
      return new KubernetesNameResolver(parts[1], parts[2], port, params, GrpcUtil.TIMER_SERVICE,
          GrpcUtil.SHARED_CHANNEL_EXECUTOR);
    } catch (NumberFormatException e) {
      throw new IllegalArgumentException("Unable to parse port number", e);
    }
  } else {
    return null;
  }
}

@Nullable
@Override
public NameResolver newNameResolver(URI targetUri, Attributes params) {
    final String scheme = targetUri.getScheme();
    if (!SCHEME_DISCOVER.equals(scheme) && !SCHEME_DIRECT.equals(scheme)) {
        return null;
    }

    final String authority = targetUri.getAuthority();
    final List<InetSocketAddress> addresses = Splitter.on(',').splitToList(authority).stream().map(host -> {
        final String[] strings = host.split(":");
        Preconditions.checkArgument(strings.length == 2, "URI should have both address and port");
        return InetSocketAddress.createUnresolved(strings[0], Integer.valueOf(strings[1]));
    }).collect(Collectors.toList());

    return new ControllerNameResolver(authority, addresses, SCHEME_DISCOVER.equals(scheme));
}

@Test
public void serverTracingSampledToLocalSpanStore() {
  ServerStreamTracer.Factory tracerFactory = censusTracing.getServerTracerFactory();
  ServerStreamTracer serverStreamTracer =
      tracerFactory.newServerStreamTracer(sampledMethod.getFullMethodName(), new Metadata());

  serverStreamTracer.filterContext(Context.ROOT);

  serverStreamTracer.serverCallStarted(
      new ServerCallInfoImpl<String, String>(sampledMethod, Attributes.EMPTY, null));

  serverStreamTracer.streamClosed(Status.CANCELLED);

  verify(spyServerSpan).end(
      EndSpanOptions.builder()
          .setStatus(io.opencensus.trace.Status.CANCELLED)
          .setSampleToLocalSpanStore(true)
          .build());
}

@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);
}

@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);
}

@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
}

@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
}

@Test
public void fail_inline() {
  final Status error = Status.FAILED_PRECONDITION.withDescription("channel not secure for creds");
  when(mockTransport.getAttributes()).thenReturn(Attributes.EMPTY);
  doAnswer(new Answer<Void>() {
      @Override
      public Void answer(InvocationOnMock invocation) throws Throwable {
        MetadataApplier applier = (MetadataApplier) invocation.getArguments()[3];
        applier.fail(error);
        return null;
      }
    }).when(mockCreds).applyRequestMetadata(same(method), any(Attributes.class),
        same(mockExecutor), any(MetadataApplier.class));

  FailingClientStream stream =
      (FailingClientStream) transport.newStream(method, origHeaders, callOptions);

  verify(mockTransport, never()).newStream(method, origHeaders, callOptions);
  assertSame(error, stream.getError());
}

@Override
protected Attributes getNameResolverParams() {
  int defaultPort;
  switch (negotiationType) {
    case PLAINTEXT:
      defaultPort = GrpcUtil.DEFAULT_PORT_PLAINTEXT;
      break;
    case TLS:
      defaultPort = GrpcUtil.DEFAULT_PORT_SSL;
      break;
    default:
      throw new AssertionError(negotiationType + " not handled");
  }
  return Attributes.newBuilder()
      .set(NameResolver.Factory.PARAMS_DEFAULT_PORT, defaultPort).build();
}

@Test
public void updateSubchannelAddresses_newAddressConnects() {
  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();

  helper.updateSubchannelAddresses(subchannel, servers.get(1));

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

/**
 * Create a server plaintext handler for gRPC.
 */
public static ProtocolNegotiator serverPlaintext() {
  return new ProtocolNegotiator() {
    @Override
    public Handler newHandler(final GrpcHttp2ConnectionHandler handler) {
      class PlaintextHandler extends ChannelHandlerAdapter implements Handler {
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
          // Set sttributes before replace to be sure we pass it before accepting any requests.
          handler.handleProtocolNegotiationCompleted(Attributes.newBuilder()
              .set(Grpc.TRANSPORT_ATTR_REMOTE_ADDR, ctx.channel().remoteAddress())
              .build());
          // Just replace this handler with the gRPC handler.
          ctx.pipeline().replace(this, null, handler);
        }

        @Override
        public AsciiString scheme() {
          return Utils.HTTP;
        }
      }

      return new PlaintextHandler();
    }
  };
}

@Override
@CheckReturnValue
protected Attributes getNameResolverParams() {
  int defaultPort;
  switch (negotiationType) {
    case PLAINTEXT:
    case PLAINTEXT_UPGRADE:
      defaultPort = GrpcUtil.DEFAULT_PORT_PLAINTEXT;
      break;
    case TLS:
      defaultPort = GrpcUtil.DEFAULT_PORT_SSL;
      break;
    default:
      throw new AssertionError(negotiationType + " not handled");
  }
  return Attributes.newBuilder()
      .set(NameResolver.Factory.PARAMS_DEFAULT_PORT, defaultPort).build();
}

@Override
protected void manualSetUp() throws Exception {
  assertNull("manualSetUp should not run more than once", handler());

  initChannel(new GrpcHttp2ServerHeadersDecoder(GrpcUtil.DEFAULT_MAX_HEADER_LIST_SIZE));

  // replace the keepAliveManager with spyKeepAliveManager
  spyKeepAliveManager =
      mock(KeepAliveManager.class, delegatesTo(handler().getKeepAliveManagerForTest()));
  handler().setKeepAliveManagerForTest(spyKeepAliveManager);

  // Simulate receipt of the connection preface
  handler().handleProtocolNegotiationCompleted(Attributes.EMPTY);
  channelRead(Http2CodecUtil.connectionPrefaceBuf());
  // Simulate receipt of initial remote settings.
  ByteBuf serializedSettings = serializeSettings(new Http2Settings());
  channelRead(serializedSettings);
}

@Test
public void parameterPropagation_overrideByCallOptions() {
  Attributes transportAttrs = Attributes.newBuilder()
      .set(ATTR_KEY, ATTR_VALUE)
      .set(CallCredentials.ATTR_AUTHORITY, "transport-override-authority")
      .set(CallCredentials.ATTR_SECURITY_LEVEL, SecurityLevel.INTEGRITY)
      .build();
  when(mockTransport.getAttributes()).thenReturn(transportAttrs);
  Executor anotherExecutor = mock(Executor.class);

  transport.newStream(method, origHeaders,
      callOptions.withAuthority("calloptions-authority").withExecutor(anotherExecutor));

  ArgumentCaptor<Attributes> attrsCaptor = ArgumentCaptor.forClass(null);
  verify(mockCreds).applyRequestMetadata(same(method), attrsCaptor.capture(),
      same(anotherExecutor), any(MetadataApplier.class));
  Attributes attrs = attrsCaptor.getValue();
  assertSame(ATTR_VALUE, attrs.get(ATTR_KEY));
  assertEquals("calloptions-authority", attrs.get(CallCredentials.ATTR_AUTHORITY));
  assertSame(SecurityLevel.INTEGRITY, attrs.get(CallCredentials.ATTR_SECURITY_LEVEL));
}

@Override
public ServerTransportListener transportCreated(final ServerTransport transport) {
  transports.add((NettyServerTransport) transport);
  return new ServerTransportListener() {
    @Override
    public void streamCreated(ServerStream stream, String method, Metadata headers) {
      EchoServerStreamListener listener = new EchoServerStreamListener(stream, method, headers);
      stream.setListener(listener);
      stream.writeHeaders(new Metadata());
      stream.request(1);
      streamListeners.add(listener);
    }

    @Override
    public Attributes transportReady(Attributes transportAttrs) {
      return transportAttrs;
    }

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

@Test
public void serviceUri() throws Exception {
  GoogleAuthLibraryCallCredentials callCredentials =
      new GoogleAuthLibraryCallCredentials(credentials);
  callCredentials.applyRequestMetadata(method,
      Attributes.newBuilder()
          .setAll(attrs)
          .set(CallCredentials.ATTR_AUTHORITY, "example.com:443")
          .build(),
      executor, applier);
  verify(credentials).getRequestMetadata(eq(new URI("https://example.com/a.service")));

  callCredentials.applyRequestMetadata(method,
      Attributes.newBuilder()
          .setAll(attrs)
          .set(CallCredentials.ATTR_AUTHORITY, "example.com:123")
          .build(),
      executor, applier);
  verify(credentials).getRequestMetadata(eq(new URI("https://example.com:123/a.service")));
}

@Test
public void decompressorNotFound() throws Exception {
  String decompressorName = "NON_EXISTENT_DECOMPRESSOR";
  createAndStartServer();
  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);
  Metadata requestHeaders = new Metadata();
  requestHeaders.put(MESSAGE_ENCODING_KEY, decompressorName);
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(
          streamTracerFactories, "Waiter/nonexist", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);

  transportListener.streamCreated(stream, "Waiter/nonexist", requestHeaders);

  verify(stream).close(statusCaptor.capture(), any(Metadata.class));
  Status status = statusCaptor.getValue();
  assertEquals(Status.Code.UNIMPLEMENTED, status.getCode());
  assertEquals("Can't find decompressor for " + decompressorName, status.getDescription());
  verifyNoMoreInteractions(stream);
}

@Test
public void applyMetadata_delayed() {
  when(mockTransport.getAttributes()).thenReturn(Attributes.EMPTY);

  // Will call applyRequestMetadata(), which is no-op.
  DelayedStream stream = (DelayedStream) transport.newStream(method, origHeaders, callOptions);

  ArgumentCaptor<MetadataApplier> applierCaptor = ArgumentCaptor.forClass(null);
  verify(mockCreds).applyRequestMetadata(same(method), any(Attributes.class),
      same(mockExecutor), applierCaptor.capture());
  verify(mockTransport, never()).newStream(method, origHeaders, callOptions);

  Metadata headers = new Metadata();
  headers.put(CREDS_KEY, CREDS_VALUE);
  applierCaptor.getValue().apply(headers);

  verify(mockTransport).newStream(method, origHeaders, callOptions);
  assertSame(mockStream, stream.getRealStream());
  assertEquals(CREDS_VALUE, origHeaders.get(CREDS_KEY));
  assertEquals(ORIG_HEADER_VALUE, origHeaders.get(ORIG_HEADER_KEY));
}

@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() {}
  };
}

@Test
public void fail_delayed() {
  when(mockTransport.getAttributes()).thenReturn(Attributes.EMPTY);

  // Will call applyRequestMetadata(), which is no-op.
  DelayedStream stream = (DelayedStream) transport.newStream(method, origHeaders, callOptions);

  ArgumentCaptor<MetadataApplier> applierCaptor = ArgumentCaptor.forClass(null);
  verify(mockCreds).applyRequestMetadata(same(method), any(Attributes.class),
      same(mockExecutor), applierCaptor.capture());

  Status error = Status.FAILED_PRECONDITION.withDescription("channel not secure for creds");
  applierCaptor.getValue().fail(error);

  verify(mockTransport, never()).newStream(method, origHeaders, callOptions);
  FailingClientStream failingStream = (FailingClientStream) stream.getRealStream();
  assertSame(error, failingStream.getError());
}

@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());
}

@Test
public void parameterPropagation_overrideByTransport() {
  Attributes transportAttrs = Attributes.newBuilder()
      .set(ATTR_KEY, ATTR_VALUE)
      .set(CallCredentials.ATTR_AUTHORITY, "transport-override-authority")
      .set(CallCredentials.ATTR_SECURITY_LEVEL, SecurityLevel.INTEGRITY)
      .build();
  when(mockTransport.getAttributes()).thenReturn(transportAttrs);

  transport.newStream(method, origHeaders, callOptions);

  ArgumentCaptor<Attributes> attrsCaptor = ArgumentCaptor.forClass(null);
  verify(mockCreds).applyRequestMetadata(same(method), attrsCaptor.capture(), same(mockExecutor),
      any(MetadataApplier.class));
  Attributes attrs = attrsCaptor.getValue();
  assertSame(ATTR_VALUE, attrs.get(ATTR_KEY));
  assertEquals("transport-override-authority", attrs.get(CallCredentials.ATTR_AUTHORITY));
  assertSame(SecurityLevel.INTEGRITY, attrs.get(CallCredentials.ATTR_SECURITY_LEVEL));
}

@Test
public void validTargetNoResovler() {
  Factory nameResolverFactory = new NameResolver.Factory() {
    @Override
    public NameResolver newNameResolver(URI targetUri, Attributes params) {
      return null;
    }

    @Override
    public String getDefaultScheme() {
      return "defaultscheme";
    }
  };
  try {
    ManagedChannelImpl.getNameResolver(
        "foo.googleapis.com:8080", nameResolverFactory, NAME_RESOLVER_PARAMS);
    fail("Should fail");
  } catch (IllegalArgumentException e) {
    // expected
  }
}

@Test
public void pickAfterResolved() throws Exception {
  final Subchannel readySubchannel = subchannels.values().iterator().next();
  loadBalancer.handleResolvedAddressGroups(servers, affinity);
  loadBalancer.handleSubchannelState(readySubchannel, ConnectivityStateInfo.forNonError(READY));

  verify(mockHelper, times(3)).createSubchannel(eagCaptor.capture(),
      any(Attributes.class));

  assertThat(eagCaptor.getAllValues()).containsAllIn(subchannels.keySet());
  for (Subchannel subchannel : subchannels.values()) {
    verify(subchannel).requestConnection();
    verify(subchannel, never()).shutdown();
  }

  verify(mockHelper, times(2))
      .updateBalancingState(stateCaptor.capture(), pickerCaptor.capture());

  assertEquals(CONNECTING, stateCaptor.getAllValues().get(0));
  assertEquals(READY, stateCaptor.getAllValues().get(1));
  assertThat(pickerCaptor.getValue().getList()).containsExactly(readySubchannel);

  verifyNoMoreInteractions(mockHelper);
}

@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);
}

@Test
public void resolve() throws Exception {
  List<InetAddress> answer1 = createAddressList(2);
  List<InetAddress> answer2 = createAddressList(1);
  String name = "foo.googleapis.com";

  DnsNameResolver resolver = newResolver(name, 81);
  mockResolver.addAnswer(answer1).addAnswer(answer2);
  resolver.start(mockListener);
  assertEquals(1, fakeExecutor.runDueTasks());
  verify(mockListener).onAddresses(resultCaptor.capture(), any(Attributes.class));
  assertEquals(name, mockResolver.invocations.poll());
  assertAnswerMatches(answer1, 81, resultCaptor.getValue());
  assertEquals(0, fakeClock.numPendingTasks());

  resolver.refresh();
  assertEquals(1, fakeExecutor.runDueTasks());
  verify(mockListener, times(2)).onAddresses(resultCaptor.capture(), any(Attributes.class));
  assertEquals(name, mockResolver.invocations.poll());
  assertAnswerMatches(answer2, 81, resultCaptor.getValue());
  assertEquals(0, fakeClock.numPendingTasks());

  resolver.shutdown();
}

@Test
public void applyMetadata_inline() {
  when(mockTransport.getAttributes()).thenReturn(Attributes.EMPTY);
  doAnswer(new Answer<Void>() {
      @Override
      public Void answer(InvocationOnMock invocation) throws Throwable {
        MetadataApplier applier = (MetadataApplier) invocation.getArguments()[3];
        Metadata headers = new Metadata();
        headers.put(CREDS_KEY, CREDS_VALUE);
        applier.apply(headers);
        return null;
      }
    }).when(mockCreds).applyRequestMetadata(same(method), any(Attributes.class),
        same(mockExecutor), any(MetadataApplier.class));

  ClientStream stream = transport.newStream(method, origHeaders, callOptions);

  verify(mockTransport).newStream(method, origHeaders, callOptions);
  assertSame(mockStream, stream);
  assertEquals(CREDS_VALUE, origHeaders.get(CREDS_KEY));
  assertEquals(ORIG_HEADER_VALUE, origHeaders.get(ORIG_HEADER_KEY));
}

@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());
}

@Test
public void forwardsNonOverridenCalls() {
  NameResolver.Factory wrappedFactory = mock(NameResolver.Factory.class);
  NameResolver mockResolver = mock(NameResolver.class);
  when(wrappedFactory.newNameResolver(any(URI.class), any(Attributes.class)))
      .thenReturn(mockResolver);
  NameResolver.Factory factory =
      new OverrideAuthorityNameResolverFactory(wrappedFactory, "override:5678");
  NameResolver overrideResolver =
      factory.newNameResolver(URI.create("dns:///localhost:443"), Attributes.EMPTY);
  assertNotNull(overrideResolver);
  NameResolver.Listener listener = mock(NameResolver.Listener.class);

  overrideResolver.start(listener);
  verify(mockResolver).start(listener);

  overrideResolver.shutdown();
  verify(mockResolver).shutdown();

  overrideResolver.refresh();
  verify(mockResolver).refresh();
}

@Test
public void methodNotFound() throws Exception {
  createAndStartServer();
  ServerTransportListener transportListener
      = transportServer.registerNewServerTransport(new SimpleServerTransport());
  transportListener.transportReady(Attributes.EMPTY);
  Metadata requestHeaders = new Metadata();
  StatsTraceContext statsTraceCtx =
      StatsTraceContext.newServerContext(
          streamTracerFactories, "Waiter/nonexist", requestHeaders);
  when(stream.statsTraceContext()).thenReturn(statsTraceCtx);
  transportListener.streamCreated(stream, "Waiter/nonexist", requestHeaders);
  verify(stream).setListener(isA(ServerStreamListener.class));
  verify(stream, atLeast(1)).statsTraceContext();

  assertEquals(1, executor.runDueTasks());
  verify(stream).close(statusCaptor.capture(), any(Metadata.class));
  Status status = statusCaptor.getValue();
  assertEquals(Status.Code.UNIMPLEMENTED, status.getCode());
  assertEquals("Method not found: Waiter/nonexist", status.getDescription());

  verify(streamTracerFactory).newServerStreamTracer(eq("Waiter/nonexist"), same(requestHeaders));
  assertNull(streamTracer.getServerCallInfo());
  assertEquals(Status.Code.UNIMPLEMENTED, statusCaptor.getValue().getCode());
}

public void refreshAffinity(Map<String, Object> affinity) {
  Attributes nameresoveCache =
      (Attributes) affinity.get(GrpcCallOptions.GRPC_NAMERESOVER_ATTRIBUTES);
  this.current_server = (SocketAddress) affinity.get(GrpcCallOptions.GRPC_CURRENT_ADDR_KEY);
  this.registry_servers = nameresoveCache.get(GrpcNameResolverProvider.REMOTE_ADDR_KEYS);
  this.listener = nameresoveCache.get(GrpcNameResolverProvider.NAMERESOVER_LISTENER);
  this.affinity = nameresoveCache;

}

private Attributes buildAttributes(GrpcURL subscribeUrl,
    Map<List<SocketAddress>, GrpcURL> addressUrlMapping) {
  Attributes.Builder builder = Attributes.newBuilder();
  if (listener != null) {
    builder.set(GrpcNameResolverProvider.NAMERESOVER_LISTENER, listener);
  }
  if (addresses.get(subscribeUrl) != null) {
    builder.set(GrpcNameResolverProvider.REMOTE_ADDR_KEYS, addresses.get(subscribeUrl));
  }
  if (!addressUrlMapping.isEmpty()) {
    builder.set(GrpcNameResolverProvider.GRPC_ADDRESS_GRPCURL_MAPPING, addressUrlMapping);
  }
  return builder.build();
}

@Nullable
@Override
public ConsulNameResolver newNameResolver(final URI targetUri, final Attributes params) {
    if (!SCHEME.equals(targetUri.getScheme())) {
        return null;
    }

    final String targetPath = checkNotNull(targetUri.getPath(), "targetPath");
    checkArgument(targetPath.startsWith("/"));

    final String serviceName = targetPath.substring(1);
    checkArgument(serviceName.length() > 0, "serviceName");

    String consulHost = targetUri.getHost();
    if (Strings.isNullOrEmpty(consulHost)) {
        consulHost = DEFAULT_HOST;
    }

    int consulPort = targetUri.getPort();
    if (consulPort == -1) {
        consulPort = DEFAULT_PORT;
    }

    final String tag = Strings.emptyToNull(targetUri.getFragment());

    final ConsulClient consulClient = ConsulClientManager.getInstance(consulHost, consulPort);

    return new ConsulNameResolver(
            consulClient /* CatalogClient */,
            consulClient /* KeyValueClient */,
            serviceName,
            Optional.ofNullable(tag),
            GrpcUtil.TIMER_SERVICE,
            GrpcUtil.SHARED_CHANNEL_EXECUTOR
    );
}

@Override
public Attributes transportReady(Attributes transportAttrs) {
    checkNotNull(transportAttrs, "transportAttrs");

    UUID sessionId = UUID.randomUUID();
    SessionLifecycleEvent event = new SessionLifecycleEvent(this, sessionId);
    sessionLifecycleEventListeners.forEach(listener -> listener.sessionStart(event));

    return Attributes.newBuilder(transportAttrs).set(TRANSPORT_ATTRIBUTES_SESSION_ID, UUID.randomUUID()).build();
}

@Override
public void transportTerminated(Attributes transportAttrs) {
    checkNotNull(transportAttrs, "transportAttrs");

    UUID sessionId = transportAttrs.get(TRANSPORT_ATTRIBUTES_SESSION_ID);
    if (sessionId != null) {
        SessionLifecycleEvent event = new SessionLifecycleEvent(this, sessionId);
        sessionLifecycleEventListeners.forEach(listener -> listener.sessionEnd(event));
    }
}

@Test
public void firstShouldFind() throws Exception {
    NameResolver fakeResolver = new FakeResolver();
    FakeResolverProvider canResolve = new FakeResolverProvider("aaa://", fakeResolver);
    FakeResolverProvider cannotResolve = new FakeResolverProvider("bbb://", null);

    NameResolver.Factory factory = FallbackResolver.startWith(canResolve).thenCheck(cannotResolve);

    assertEquals(fakeResolver, factory.newNameResolver(new URI("aaa://foo"), Attributes.EMPTY));
}