Java 类io.grpc.LoadBalancer.Helper 实例源码

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

@Test
public void getState_withRequestConnect() {
  createChannel(
      new FakeNameResolverFactory(false), NO_INTERCEPTOR, false /* requestConnection */,
      ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE);

  assertEquals(IDLE, channel.getState(false));
  verify(mockLoadBalancerFactory, never()).newLoadBalancer(any(Helper.class));

  // call getState() with requestConnection = true
  assertEquals(IDLE, channel.getState(true));
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  helper = helperCaptor.getValue();

  helper.updateBalancingState(CONNECTING, mockPicker);
  assertEquals(CONNECTING, channel.getState(false));
  assertEquals(CONNECTING, channel.getState(true));
  verifyNoMoreInteractions(mockLoadBalancerFactory);
}

@Test
public void idleTimeoutAndReconnect() {
  long idleTimeoutMillis = 2000L;
  createChannel(
      new FakeNameResolverFactory(true), NO_INTERCEPTOR, true /* request connection*/,
      idleTimeoutMillis);

  timer.forwardNanos(TimeUnit.MILLISECONDS.toNanos(idleTimeoutMillis));
  assertEquals(IDLE, channel.getState(true /* request connection */));

  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(Helper.class);
  // Two times of requesting connection will create loadBalancer twice.
  verify(mockLoadBalancerFactory, times(2)).newLoadBalancer(helperCaptor.capture());
  Helper helper2 = helperCaptor.getValue();

  // Updating on the old helper (whose balancer has been shutdown) does not change the channel
  // state.
  helper.updateBalancingState(CONNECTING, mockPicker);
  assertEquals(IDLE, channel.getState(false));

  helper2.updateBalancingState(CONNECTING, mockPicker);
  assertEquals(CONNECTING, channel.getState(false));
}

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

@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 updateOobChannelAddresses_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();
  ManagedChannel oobChannel = helper.createOobChannel(servers.get(0), "localhost");

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

  helper.updateOobChannelAddresses(oobChannel, servers.get(1));

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

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

GrpclbState(
    Helper helper,
    TimeProvider time,
    ScheduledExecutorService timerService,
    LogId logId) {
  this.helper = checkNotNull(helper, "helper");
  this.time = checkNotNull(time, "time provider");
  this.timerService = checkNotNull(timerService, "timerService");
  this.serviceName = checkNotNull(helper.getAuthority(), "helper returns null authority");
  this.logId = checkNotNull(logId, "logId");
}

private List<Subchannel> fallbackTestVerifyUseOfBalancerBackendLists(
    InOrder inOrder, Helper helper, List<ServerEntry> servers) {
  ArrayList<EquivalentAddressGroup> addrs = new ArrayList<EquivalentAddressGroup>();
  ArrayList<String> tokens = new ArrayList<String>();
  for (ServerEntry server : servers) {
    addrs.add(new EquivalentAddressGroup(server.addr));
    tokens.add(server.token);
  }
  return fallbackTestVerifyUseOfBackendLists(inOrder, helper, addrs, tokens);
}

@Before
public void setUp() throws Exception {
  MockitoAnnotations.initMocks(this);
  expectedUri = new URI(target);
  when(mockLoadBalancerFactory.newLoadBalancer(any(Helper.class))).thenReturn(mockLoadBalancer);
  transports = TestUtils.captureTransports(mockTransportFactory);
  when(mockTransportFactory.getScheduledExecutorService())
      .thenReturn(timer.getScheduledExecutorService());
  when(executorPool.getObject()).thenReturn(executor.getScheduledExecutorService());
  when(oobExecutorPool.getObject()).thenReturn(oobExecutor.getScheduledExecutorService());
}

@Test
public void nameResolverReturnsEmptySubLists() {
  String errorDescription = "NameResolver returned an empty list";

  // Pass a FakeNameResolverFactory with an empty list
  createChannel(new FakeNameResolverFactory(), NO_INTERCEPTOR);

  // LoadBalancer received the error
  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
  verify(mockLoadBalancer).handleNameResolutionError(statusCaptor.capture());
  Status status = statusCaptor.getValue();
  assertSame(Status.Code.UNAVAILABLE, status.getCode());
  assertEquals(errorDescription, status.getDescription());
}

@Test
public void channelsAndSubchannels_instrumented_state() throws Exception {
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  helper = helperCaptor.getValue();

  assertEquals(IDLE, getStats(channel).state);
  helper.updateBalancingState(CONNECTING, mockPicker);
  assertEquals(CONNECTING, getStats(channel).state);

  AbstractSubchannel subchannel =
      (AbstractSubchannel) helper.createSubchannel(addressGroup, Attributes.EMPTY);

  assertEquals(IDLE, getStats(subchannel).state);
  subchannel.requestConnection();
  assertEquals(CONNECTING, getStats(subchannel).state);

  MockClientTransportInfo transportInfo = transports.poll();

  assertEquals(CONNECTING, getStats(subchannel).state);
  transportInfo.listener.transportReady();
  assertEquals(READY, getStats(subchannel).state);

  assertEquals(CONNECTING, getStats(channel).state);
  helper.updateBalancingState(READY, mockPicker);
  assertEquals(READY, getStats(channel).state);

  channel.shutdownNow();
  assertEquals(SHUTDOWN, getStats(channel).state);
  assertEquals(SHUTDOWN, getStats(subchannel).state);
}

@Test
public void newCallExitsIdleness() throws Exception {
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());

  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));

  verify(mockNameResolver).start(nameResolverListenerCaptor.capture());
  // Simulate new address resolved to make sure the LoadBalancer is correctly linked to
  // the NameResolver.
  nameResolverListenerCaptor.getValue().onAddresses(servers, Attributes.EMPTY);
  verify(mockLoadBalancer).handleResolvedAddressGroups(servers, Attributes.EMPTY);
}

@Test
public void newCallRefreshesIdlenessTimer() throws Exception {
  // First call to exit the initial idleness, then immediately cancel the call.
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());
  call.cancel("For testing", null);

  // Verify that we have exited the idle mode
  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
  assertFalse(channel.inUseStateAggregator.isInUse());

  // Move closer to idleness, but not yet.
  timer.forwardTime(IDLE_TIMEOUT_SECONDS - 1, TimeUnit.SECONDS);
  verify(mockLoadBalancer, never()).shutdown();
  assertFalse(channel.inUseStateAggregator.isInUse());

  // A new call would refresh the timer
  call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());
  call.cancel("For testing", null);
  assertFalse(channel.inUseStateAggregator.isInUse());

  // ... so that passing the same length of time will not trigger idle mode
  timer.forwardTime(IDLE_TIMEOUT_SECONDS - 1, TimeUnit.SECONDS);
  verify(mockLoadBalancer, never()).shutdown();
  assertFalse(channel.inUseStateAggregator.isInUse());

  // ... until the time since last call has reached the timeout
  timer.forwardTime(1, TimeUnit.SECONDS);
  verify(mockLoadBalancer).shutdown();
  assertFalse(channel.inUseStateAggregator.isInUse());

  // Drain the app executor, which runs the call listeners
  verify(mockCallListener, never()).onClose(any(Status.class), any(Metadata.class));
  assertEquals(2, executor.runDueTasks());
  verify(mockCallListener, times(2)).onClose(any(Status.class), any(Metadata.class));
}

@Override
public LoadBalancer newLoadBalancer(Helper helper) {
  return new GrpcRoundRobinLoadBalancer(helper);
}

GrpcRoundRobinLoadBalancer(Helper helper) {
  this.helper = checkNotNull(helper, "helper");
}

private List<Subchannel> fallbackTestVerifyUseOfFallbackBackendLists(
    InOrder inOrder, Helper helper, List<EquivalentAddressGroup> addrs) {
  return fallbackTestVerifyUseOfBackendLists(inOrder, helper, addrs, null);
}

private void createChannel(
    NameResolver.Factory nameResolverFactory, List<ClientInterceptor> interceptors,
    boolean requestConnection, long idleTimeoutMillis) {
  class Builder extends AbstractManagedChannelImplBuilder<Builder> {
    Builder(String target) {
      super(target);
    }

    @Override protected ClientTransportFactory buildTransportFactory() {
      throw new UnsupportedOperationException();
    }

    @Override protected Attributes getNameResolverParams() {
      return NAME_RESOLVER_PARAMS;
    }

    @Override public Builder usePlaintext(boolean b) {
      throw new UnsupportedOperationException();
    }
  }

  Builder builder = new Builder(target)
      .nameResolverFactory(nameResolverFactory)
      .loadBalancerFactory(mockLoadBalancerFactory)
      .userAgent(userAgent);
  builder.executorPool = executorPool;
  builder.idleTimeoutMillis = idleTimeoutMillis;
  builder.binlogProvider = binlogProvider;
  checkState(channel == null);
  channel = new ManagedChannelImpl(
      builder, mockTransportFactory, new FakeBackoffPolicyProvider(),
      oobExecutorPool, timer.getStopwatchSupplier(), interceptors, GrpcUtil.NOOP_PROXY_DETECTOR,
      channelStatsFactory);

  if (requestConnection) {
    // Force-exit the initial idle-mode
    channel.exitIdleMode();
    assertEquals(
        idleTimeoutMillis == ManagedChannelImpl.IDLE_TIMEOUT_MILLIS_DISABLE ? 0 : 1,
        timer.numPendingTasks());

    ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
    verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
    helper = helperCaptor.getValue();
  }
}

@Test
public void noMoreCallbackAfterLoadBalancerShutdown() {
  FakeNameResolverFactory nameResolverFactory = new FakeNameResolverFactory(true);
  Status resolutionError = Status.UNAVAILABLE.withDescription("Resolution failed");
  createChannel(nameResolverFactory, NO_INTERCEPTOR);

  FakeNameResolverFactory.FakeNameResolver resolver = nameResolverFactory.resolvers.get(0);
  verify(mockLoadBalancerFactory).newLoadBalancer(any(Helper.class));
  verify(mockLoadBalancer).handleResolvedAddressGroups(
      eq(Arrays.asList(addressGroup)), eq(Attributes.EMPTY));

  Subchannel subchannel1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  Subchannel subchannel2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel1.requestConnection();
  subchannel2.requestConnection();
  verify(mockTransportFactory, times(2)).newClientTransport(
      any(SocketAddress.class), any(String.class), any(String.class), any(ProxyParameters.class));
  MockClientTransportInfo transportInfo1 = transports.poll();
  MockClientTransportInfo transportInfo2 = transports.poll();

  // LoadBalancer receives all sorts of callbacks
  transportInfo1.listener.transportReady();
  verify(mockLoadBalancer, times(2))
      .handleSubchannelState(same(subchannel1), stateInfoCaptor.capture());
  assertSame(CONNECTING, stateInfoCaptor.getAllValues().get(0).getState());
  assertSame(READY, stateInfoCaptor.getAllValues().get(1).getState());

  verify(mockLoadBalancer)
      .handleSubchannelState(same(subchannel2), stateInfoCaptor.capture());
  assertSame(CONNECTING, stateInfoCaptor.getValue().getState());

  resolver.listener.onError(resolutionError);
  verify(mockLoadBalancer).handleNameResolutionError(resolutionError);

  verifyNoMoreInteractions(mockLoadBalancer);

  channel.shutdown();
  verify(mockLoadBalancer).shutdown();

  // No more callback should be delivered to LoadBalancer after it's shut down
  transportInfo2.listener.transportReady();
  resolver.listener.onError(resolutionError);
  resolver.resolved();
  verifyNoMoreInteractions(mockLoadBalancer);
}

@Before
public void setUp() {
  MockitoAnnotations.initMocks(this);
  when(mockLoadBalancerFactory.newLoadBalancer(any(Helper.class))).thenReturn(mockLoadBalancer);
  when(mockNameResolver.getServiceAuthority()).thenReturn(AUTHORITY);
  when(mockNameResolverFactory
      .newNameResolver(any(URI.class), any(Attributes.class)))
      .thenReturn(mockNameResolver);
  when(mockTransportFactory.getScheduledExecutorService())
      .thenReturn(timer.getScheduledExecutorService());

  class Builder extends AbstractManagedChannelImplBuilder<Builder> {
    Builder(String target) {
      super(target);
    }

    @Override protected ClientTransportFactory buildTransportFactory() {
      throw new UnsupportedOperationException();
    }

    @Override public Builder usePlaintext(boolean b) {
      throw new UnsupportedOperationException();
    }
  }

  Builder builder = new Builder("fake://target")
      .nameResolverFactory(mockNameResolverFactory)
      .loadBalancerFactory(mockLoadBalancerFactory)
      .idleTimeout(IDLE_TIMEOUT_SECONDS, TimeUnit.SECONDS)
      .userAgent(USER_AGENT);
  builder.executorPool = executorPool;
  channel = new ManagedChannelImpl(
      builder, mockTransportFactory, new FakeBackoffPolicyProvider(),
      oobExecutorPool, timer.getStopwatchSupplier(),
      Collections.<ClientInterceptor>emptyList(),
      GrpcUtil.NOOP_PROXY_DETECTOR, CallTracer.getDefaultFactory());
  newTransports = TestUtils.captureTransports(mockTransportFactory);

  for (int i = 0; i < 2; i++) {
    ArrayList<SocketAddress> addrs = Lists.newArrayList();
    for (int j = 0; j < 2; j++) {
      addrs.add(new FakeSocketAddress("servergroup" + i + "server" + j));
    }
    servers.add(new EquivalentAddressGroup(addrs));
  }
  verify(mockNameResolverFactory).newNameResolver(any(URI.class), any(Attributes.class));
  // Verify the initial idleness
  verify(mockLoadBalancerFactory, never()).newLoadBalancer(any(Helper.class));
  verify(mockTransportFactory, never()).newClientTransport(
      any(SocketAddress.class), anyString(), anyString(), any(ProxyParameters.class));
  verify(mockNameResolver, never()).start(any(NameResolver.Listener.class));
}

@Test
public void realTransportsHoldsOffIdleness() throws Exception {
  final EquivalentAddressGroup addressGroup = servers.get(1);

  // Start a call, which goes to delayed transport
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());

  // Verify that we have exited the idle mode
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();
  assertTrue(channel.inUseStateAggregator.isInUse());

  // Assume LoadBalancer has received an address, then create a subchannel.
  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel.requestConnection();
  MockClientTransportInfo t0 = newTransports.poll();
  t0.listener.transportReady();

  SubchannelPicker mockPicker = mock(SubchannelPicker.class);
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel));
  helper.updateBalancingState(READY, mockPicker);
  // Delayed transport creates real streams in the app executor
  executor.runDueTasks();

  // Delayed transport exits in-use, while real transport has not entered in-use yet.
  assertFalse(channel.inUseStateAggregator.isInUse());

  // Now it's in-use
  t0.listener.transportInUse(true);
  assertTrue(channel.inUseStateAggregator.isInUse());

  // As long as the transport is in-use, the channel won't go idle.
  timer.forwardTime(IDLE_TIMEOUT_SECONDS * 2, TimeUnit.SECONDS);
  assertTrue(channel.inUseStateAggregator.isInUse());

  t0.listener.transportInUse(false);
  assertFalse(channel.inUseStateAggregator.isInUse());
  // And allow the channel to go idle.
  timer.forwardTime(IDLE_TIMEOUT_SECONDS - 1, TimeUnit.SECONDS);
  verify(mockLoadBalancer, never()).shutdown();
  timer.forwardTime(1, TimeUnit.SECONDS);
  verify(mockLoadBalancer).shutdown();
}

@Test
public void oobTransportDoesNotAffectIdleness() {
  // Start a call, which goes to delayed transport
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());

  // Verify that we have exited the idle mode
  ArgumentCaptor<Helper> helperCaptor = ArgumentCaptor.forClass(null);
  verify(mockLoadBalancerFactory).newLoadBalancer(helperCaptor.capture());
  Helper helper = helperCaptor.getValue();

  // Fail the RPC
  SubchannelPicker failingPicker = mock(SubchannelPicker.class);
  when(failingPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withError(Status.UNAVAILABLE));
  helper.updateBalancingState(TRANSIENT_FAILURE, failingPicker);
  executor.runDueTasks();
  verify(mockCallListener).onClose(same(Status.UNAVAILABLE), any(Metadata.class));

  // ... so that the channel resets its in-use state
  assertFalse(channel.inUseStateAggregator.isInUse());

  // Now make an RPC on an OOB channel
  ManagedChannel oob = helper.createOobChannel(servers.get(0), "oobauthority");
  verify(mockTransportFactory, never())
      .newClientTransport(any(SocketAddress.class), same("oobauthority"), same(USER_AGENT),
          same(NO_PROXY));
  ClientCall<String, Integer> oobCall = oob.newCall(method, CallOptions.DEFAULT);
  oobCall.start(mockCallListener2, new Metadata());
  verify(mockTransportFactory)
      .newClientTransport(any(SocketAddress.class), same("oobauthority"), same(USER_AGENT),
          same(NO_PROXY));
  MockClientTransportInfo oobTransportInfo = newTransports.poll();
  assertEquals(0, newTransports.size());
  // The OOB transport reports in-use state
  oobTransportInfo.listener.transportInUse(true);

  // But it won't stop the channel from going idle
  verify(mockLoadBalancer, never()).shutdown();
  timer.forwardTime(IDLE_TIMEOUT_SECONDS, TimeUnit.SECONDS);
  verify(mockLoadBalancer).shutdown();
}