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

void handleSubchannelStateChange(final ConnectivityStateInfo newState) {
  switch (newState.getState()) {
    case READY:
    case IDLE:
      delayedTransport.reprocess(subchannelPicker);
      break;
    case TRANSIENT_FAILURE:
      delayedTransport.reprocess(new SubchannelPicker() {
          final PickResult errorResult = PickResult.withError(newState.getStatus());

          @Override
          public PickResult pickSubchannel(PickSubchannelArgs args) {
            return errorResult;
          }
        });
      break;
    default:
      // Do nothing
  }
}

@Test
public void reprocess_NoPendingStream() {
  SubchannelPicker picker = mock(SubchannelPicker.class);
  AbstractSubchannel subchannel = mock(AbstractSubchannel.class);
  when(subchannel.obtainActiveTransport()).thenReturn(mockRealTransport);
  when(picker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
      PickResult.withSubchannel(subchannel));
  when(mockRealTransport.newStream(any(MethodDescriptor.class), any(Metadata.class),
          any(CallOptions.class))).thenReturn(mockRealStream);
  delayedTransport.reprocess(picker);
  verifyNoMoreInteractions(picker);
  verifyNoMoreInteractions(transportListener);

  // Though picker was not originally used, it will be saved and serve future streams.
  ClientStream stream = delayedTransport.newStream(method, headers, CallOptions.DEFAULT);
  verify(picker).pickSubchannel(new PickSubchannelArgsImpl(method, headers, CallOptions.DEFAULT));
  verify(subchannel).obtainActiveTransport();
  assertSame(mockRealStream, stream);
}

@Override
public PickResult pickSubchannel(PickSubchannelArgs args) {
  Map<String, Object> affinity =
      args.getCallOptions().getOption(GrpcCallOptions.CALLOPTIONS_CUSTOME_KEY);
  GrpcURL refUrl = (GrpcURL) affinity.get(GrpcCallOptions.GRPC_REF_URL);
  if (size > 0) {
    Subchannel subchannel = nextSubchannel(refUrl);
    affinity.put(GrpcCallOptions.GRPC_NAMERESOVER_ATTRIBUTES, nameResovleCache);
    return PickResult.withSubchannel(subchannel);
  }
  if (status != null) {
    return PickResult.withError(status);
  }
  return PickResult.withNoResult();
}

@Override
public PickResult pickSubchannel(PickSubchannelArgs args) {
  synchronized (pickList) {
    // Two-level round-robin.
    // First round-robin on dropList. If a drop entry is selected, request will be dropped.  If
    // a non-drop entry is selected, then round-robin on pickList.  This makes sure requests are
    // dropped at the same proportion as the drop entries appear on the round-robin list from
    // the balancer, while only READY backends (that make up pickList) are selected for the
    // non-drop cases.
    if (!dropList.isEmpty()) {
      DropEntry drop = dropList.get(dropIndex);
      dropIndex++;
      if (dropIndex == dropList.size()) {
        dropIndex = 0;
      }
      if (drop != null) {
        return drop.picked();
      }
    }

    RoundRobinEntry pick = pickList.get(pickIndex);
    pickIndex++;
    if (pickIndex == pickList.size()) {
      pickIndex = 0;
    }
    return pick.picked(args.getHeaders());
  }
}

@Test
public void roundRobinPickerNoDrop() {
  GrpclbClientLoadRecorder loadRecorder = new GrpclbClientLoadRecorder(timeProvider);
  Subchannel subchannel = mock(Subchannel.class);
  BackendEntry b1 = new BackendEntry(subchannel, loadRecorder, "LBTOKEN0001");
  BackendEntry b2 = new BackendEntry(subchannel, loadRecorder, "LBTOKEN0002");

  List<BackendEntry> pickList = Arrays.asList(b1, b2);
  RoundRobinPicker picker = new RoundRobinPicker(Collections.<DropEntry>emptyList(), pickList);

  PickSubchannelArgs args1 = mock(PickSubchannelArgs.class);
  Metadata headers1 = new Metadata();
  // The existing token on the headers will be replaced
  headers1.put(GrpclbConstants.TOKEN_METADATA_KEY, "LBTOKEN__OLD");
  when(args1.getHeaders()).thenReturn(headers1);
  assertSame(b1.result, picker.pickSubchannel(args1));
  verify(args1).getHeaders();
  assertThat(headers1.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0001");

  PickSubchannelArgs args2 = mock(PickSubchannelArgs.class);
  Metadata headers2 = new Metadata();
  when(args2.getHeaders()).thenReturn(headers2);
  assertSame(b2.result, picker.pickSubchannel(args2));
  verify(args2).getHeaders();
  assertThat(headers2.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0002");

  PickSubchannelArgs args3 = mock(PickSubchannelArgs.class);
  Metadata headers3 = new Metadata();
  when(args3.getHeaders()).thenReturn(headers3);
  assertSame(b1.result, picker.pickSubchannel(args3));
  verify(args3).getHeaders();
  assertThat(headers3.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0001");

  verify(subchannel, never()).getAttributes();
}

@Test
public void abundantInitialResponse() {
  Metadata headers = new Metadata();
  PickSubchannelArgs args = mock(PickSubchannelArgs.class);
  when(args.getHeaders()).thenReturn(headers);

  List<EquivalentAddressGroup> grpclbResolutionList = createResolvedServerAddresses(true);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);
  assertEquals(1, fakeOobChannels.size());
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());
  StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue();

  // Simulate LB initial response
  assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  lbResponseObserver.onNext(buildInitialResponse(1983));

  // Load reporting task is scheduled
  assertEquals(1, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  FakeClock.ScheduledTask scheduledTask = fakeClock.getPendingTasks().iterator().next();
  assertEquals(1983, scheduledTask.getDelay(TimeUnit.MILLISECONDS));

  // Simulate an abundant LB initial response, with a different report interval
  lbResponseObserver.onNext(buildInitialResponse(9097));
  // It doesn't affect load-reporting at all
  assertThat(fakeClock.getPendingTasks(LOAD_REPORTING_TASK_FILTER))
      .containsExactly(scheduledTask);
  assertEquals(1983, scheduledTask.getDelay(TimeUnit.MILLISECONDS));
}

@Override
public ClientTransport get(PickSubchannelArgs args) {
  SubchannelPicker pickerCopy = subchannelPicker;
  if (shutdown.get()) {
    // If channel is shut down, delayedTransport is also shut down which will fail the stream
    // properly.
    return delayedTransport;
  }
  if (pickerCopy == null) {
    channelExecutor.executeLater(new Runnable() {
        @Override
        public void run() {
          exitIdleMode();
        }
      }).drain();
    return delayedTransport;
  }
  // There is no need to reschedule the idle timer here.
  //
  // pickerCopy != null, which means idle timer has not expired when this method starts.
  // Even if idle timer expires right after we grab pickerCopy, and it shuts down LoadBalancer
  // which calls Subchannel.shutdown(), the InternalSubchannel will be actually shutdown after
  // SUBCHANNEL_SHUTDOWN_DELAY_SECONDS, which gives the caller time to start RPC on it.
  //
  // In most cases the idle timer is scheduled to fire after the transport has created the
  // stream, which would have reported in-use state to the channel that would have cancelled
  // the idle timer.
  PickResult pickResult = pickerCopy.pickSubchannel(args);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(
      pickResult, args.getCallOptions().isWaitForReady());
  if (transport != null) {
    return transport;
  }
  return delayedTransport;
}

/**
 * Caller must call {@code channelExecutor.drain()} outside of lock because this method may
 * schedule tasks on channelExecutor.
 */
@GuardedBy("lock")
private PendingStream createPendingStream(PickSubchannelArgs args) {
  PendingStream pendingStream = new PendingStream(args);
  pendingStreams.add(pendingStream);
  if (getPendingStreamsCount() == 1) {
    channelExecutor.executeLater(reportTransportInUse);
  }
  return pendingStream;
}

@Override
public ClientTransport get(PickSubchannelArgs args) {
  // delayed transport's newStream() always acquires a lock, but concurrent performance doesn't
  // matter here because OOB communication should be sparse, and it's not on application RPC's
  // critical path.
  return delayedTransport;
}

@Override
public PickResult pickSubchannel(PickSubchannelArgs args) {
  if (list.size() > 0) {
    return PickResult.withSubchannel(nextSubchannel());
  }

  if (status != null) {
    return PickResult.withError(status);
  }

  return PickResult.withNoResult();
}

private void subtestFailRpcFromBalancer(boolean waitForReady, boolean drop, boolean shouldFail) {
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  // This call will be buffered by the channel, thus involve delayed transport
  CallOptions callOptions = CallOptions.DEFAULT;
  if (waitForReady) {
    callOptions = callOptions.withWaitForReady();
  } else {
    callOptions = callOptions.withoutWaitForReady();
  }
  ClientCall<String, Integer> call1 = channel.newCall(method, callOptions);
  call1.start(mockCallListener, new Metadata());

  SubchannelPicker picker = mock(SubchannelPicker.class);
  Status status = Status.UNAVAILABLE.withDescription("for test");

  when(picker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(drop ? PickResult.withDrop(status) : PickResult.withError(status));
  helper.updateBalancingState(READY, picker);

  executor.runDueTasks();
  if (shouldFail) {
    verify(mockCallListener).onClose(same(status), any(Metadata.class));
  } else {
    verifyZeroInteractions(mockCallListener);
  }

  // This call doesn't involve delayed transport
  ClientCall<String, Integer> call2 = channel.newCall(method, callOptions);
  call2.start(mockCallListener2, new Metadata());

  executor.runDueTasks();
  if (shouldFail) {
    verify(mockCallListener2).onClose(same(status), any(Metadata.class));
  } else {
    verifyZeroInteractions(mockCallListener2);
  }
}

@Test
public void pickerReturnsStreamTracer_noDelay() {
  ClientStream mockStream = mock(ClientStream.class);
  ClientStreamTracer.Factory factory1 = mock(ClientStreamTracer.Factory.class);
  ClientStreamTracer.Factory factory2 = mock(ClientStreamTracer.Factory.class);
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);
  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel.requestConnection();
  MockClientTransportInfo transportInfo = transports.poll();
  transportInfo.listener.transportReady();
  ClientTransport mockTransport = transportInfo.transport;
  when(mockTransport.newStream(
          any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
      .thenReturn(mockStream);

  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
      PickResult.withSubchannel(subchannel, factory2));
  helper.updateBalancingState(READY, mockPicker);

  CallOptions callOptions = CallOptions.DEFAULT.withStreamTracerFactory(factory1);
  ClientCall<String, Integer> call = channel.newCall(method, callOptions);
  call.start(mockCallListener, new Metadata());

  verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
  verify(mockTransport).newStream(same(method), any(Metadata.class), callOptionsCaptor.capture());
  assertEquals(
      Arrays.asList(factory1, factory2),
      callOptionsCaptor.getValue().getStreamTracerFactories());
  // The factories are safely not stubbed because we do not expect any usage of them.
  verifyZeroInteractions(factory1);
  verifyZeroInteractions(factory2);
}

@Test
public void pickerReturnsStreamTracer_delayed() {
  ClientStream mockStream = mock(ClientStream.class);
  ClientStreamTracer.Factory factory1 = mock(ClientStreamTracer.Factory.class);
  ClientStreamTracer.Factory factory2 = mock(ClientStreamTracer.Factory.class);
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  CallOptions callOptions = CallOptions.DEFAULT.withStreamTracerFactory(factory1);
  ClientCall<String, Integer> call = channel.newCall(method, callOptions);
  call.start(mockCallListener, new Metadata());

  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel.requestConnection();
  MockClientTransportInfo transportInfo = transports.poll();
  transportInfo.listener.transportReady();
  ClientTransport mockTransport = transportInfo.transport;
  when(mockTransport.newStream(
          any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
      .thenReturn(mockStream);
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class))).thenReturn(
      PickResult.withSubchannel(subchannel, factory2));

  helper.updateBalancingState(READY, mockPicker);
  assertEquals(1, executor.runDueTasks());

  verify(mockPicker).pickSubchannel(any(PickSubchannelArgs.class));
  verify(mockTransport).newStream(same(method), any(Metadata.class), callOptionsCaptor.capture());
  assertEquals(
      Arrays.asList(factory1, factory2),
      callOptionsCaptor.getValue().getStreamTracerFactories());
  // The factories are safely not stubbed because we do not expect any usage of them.
  verifyZeroInteractions(factory1);
  verifyZeroInteractions(factory2);
}

@Test
public void updateBalancingStateDoesUpdatePicker() {
  ClientStream mockStream = mock(ClientStream.class);
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  call.start(mockCallListener, new Metadata());

  // Make the transport available with subchannel2
  Subchannel subchannel1 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  Subchannel subchannel2 = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel2.requestConnection();

  MockClientTransportInfo transportInfo = transports.poll();
  ConnectionClientTransport mockTransport = transportInfo.transport;
  ManagedClientTransport.Listener transportListener = transportInfo.listener;
  when(mockTransport.newStream(same(method), any(Metadata.class), any(CallOptions.class)))
      .thenReturn(mockStream);
  transportListener.transportReady();

  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel1));
  helper.updateBalancingState(READY, mockPicker);

  executor.runDueTasks();
  verify(mockTransport, never())
      .newStream(any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class));
  verify(mockStream, never()).start(any(ClientStreamListener.class));


  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel2));
  helper.updateBalancingState(READY, mockPicker);

  executor.runDueTasks();
  verify(mockTransport).newStream(same(method), any(Metadata.class), any(CallOptions.class));
  verify(mockStream).start(any(ClientStreamListener.class));
}

@Test
public void channelStat_callEndSuccess() throws Exception {
  // set up
  Metadata headers = new Metadata();
  ClientStream mockStream = mock(ClientStream.class);
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  // Start a call with a call executor
  CallOptions options =
      CallOptions.DEFAULT.withExecutor(executor.getScheduledExecutorService());
  ClientCall<String, Integer> call = channel.newCall(method, options);
  call.start(mockCallListener, headers);

  // Make the transport available
  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  subchannel.requestConnection();
  MockClientTransportInfo transportInfo = transports.poll();
  ConnectionClientTransport mockTransport = transportInfo.transport;
  ManagedClientTransport.Listener transportListener = transportInfo.listener;
  when(mockTransport.newStream(same(method), same(headers), any(CallOptions.class)))
      .thenReturn(mockStream);
  transportListener.transportReady();
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel));
  helper.updateBalancingState(READY, mockPicker);

  executor.runDueTasks();
  verify(mockStream).start(streamListenerCaptor.capture());
  // end set up

  // the actual test
  ClientStreamListener streamListener = streamListenerCaptor.getValue();
  call.halfClose();
  assertEquals(0, getStats(channel).callsSucceeded);
  assertEquals(0, getStats(channel).callsFailed);
  streamListener.closed(Status.OK, new Metadata());
  executor.runDueTasks();
  assertEquals(1, getStats(channel).callsSucceeded);
  assertEquals(0, getStats(channel).callsFailed);
}

@Before public void setUp() {
  MockitoAnnotations.initMocks(this);
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(mockSubchannel));
  when(mockSubchannel.obtainActiveTransport()).thenReturn(mockRealTransport);
  when(mockRealTransport.newStream(same(method), same(headers), same(callOptions)))
      .thenReturn(mockRealStream);
  when(mockRealTransport2.newStream(same(method2), same(headers2), same(callOptions2)))
      .thenReturn(mockRealStream2);
  delayedTransport.start(transportListener);
}

@Test
public void roundRobinPickerWithDrop() {
  assertTrue(DROP_PICK_RESULT.isDrop());
  GrpclbClientLoadRecorder loadRecorder = new GrpclbClientLoadRecorder(timeProvider);
  Subchannel subchannel = mock(Subchannel.class);
  // 1 out of 2 requests are to be dropped
  DropEntry d = new DropEntry(loadRecorder, "LBTOKEN0003");
  List<DropEntry> dropList = Arrays.asList(null, d);

  BackendEntry b1 = new BackendEntry(subchannel, loadRecorder, "LBTOKEN0001");
  BackendEntry b2 = new BackendEntry(subchannel, loadRecorder, "LBTOKEN0002");
  List<BackendEntry> pickList = Arrays.asList(b1, b2);
  RoundRobinPicker picker = new RoundRobinPicker(dropList, pickList);

  // dropList[0], pickList[0]
  PickSubchannelArgs args1 = mock(PickSubchannelArgs.class);
  Metadata headers1 = new Metadata();
  headers1.put(GrpclbConstants.TOKEN_METADATA_KEY, "LBTOKEN__OLD");
  when(args1.getHeaders()).thenReturn(headers1);
  assertSame(b1.result, picker.pickSubchannel(args1));
  verify(args1).getHeaders();
  assertThat(headers1.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0001");

  // dropList[1]: drop
  PickSubchannelArgs args2 = mock(PickSubchannelArgs.class);
  Metadata headers2 = new Metadata();
  when(args2.getHeaders()).thenReturn(headers2);
  assertSame(DROP_PICK_RESULT, picker.pickSubchannel(args2));
  verify(args2, never()).getHeaders();

  // dropList[0], pickList[1]
  PickSubchannelArgs args3 = mock(PickSubchannelArgs.class);
  Metadata headers3 = new Metadata();
  when(args3.getHeaders()).thenReturn(headers3);
  assertSame(b2.result, picker.pickSubchannel(args3));
  verify(args3).getHeaders();
  assertThat(headers3.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0002");

  // dropList[1]: drop
  PickSubchannelArgs args4 = mock(PickSubchannelArgs.class);
  Metadata headers4 = new Metadata();
  when(args4.getHeaders()).thenReturn(headers4);
  assertSame(DROP_PICK_RESULT, picker.pickSubchannel(args4));
  verify(args4, never()).getHeaders();

  // dropList[0], pickList[0]
  PickSubchannelArgs args5 = mock(PickSubchannelArgs.class);
  Metadata headers5 = new Metadata();
  when(args5.getHeaders()).thenReturn(headers5);
  assertSame(b1.result, picker.pickSubchannel(args5));
  verify(args5).getHeaders();
  assertThat(headers5.getAll(GrpclbConstants.TOKEN_METADATA_KEY)).containsExactly("LBTOKEN0001");

  verify(subchannel, never()).getAttributes();
}

@Test
public void raceBetweenLoadReportingAndLbStreamClosure() {
  Metadata headers = new Metadata();
  PickSubchannelArgs args = mock(PickSubchannelArgs.class);
  when(args.getHeaders()).thenReturn(headers);

  List<EquivalentAddressGroup> grpclbResolutionList = createResolvedServerAddresses(true);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);
  assertEquals(1, fakeOobChannels.size());
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());
  StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue();
  assertEquals(1, lbRequestObservers.size());
  StreamObserver<LoadBalanceRequest> lbRequestObserver = lbRequestObservers.poll();
  InOrder inOrder = inOrder(lbRequestObserver);

  inOrder.verify(lbRequestObserver).onNext(
      eq(LoadBalanceRequest.newBuilder().setInitialRequest(
              InitialLoadBalanceRequest.newBuilder().setName(SERVICE_AUTHORITY).build())
          .build()));

  // Simulate receiving LB response
  assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  lbResponseObserver.onNext(buildInitialResponse(1983));

  // Load reporting task is scheduled
  assertEquals(1, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
  FakeClock.ScheduledTask scheduledTask = fakeClock.getPendingTasks().iterator().next();
  assertEquals(1983, scheduledTask.getDelay(TimeUnit.MILLISECONDS));

  // Close lbStream
  lbResponseObserver.onCompleted();

  // Reporting task cancelled
  assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));

  // Simulate a race condition where the task has just started when its cancelled
  scheduledTask.command.run();

  // No report sent. No new task scheduled
  inOrder.verify(lbRequestObserver, never()).onNext(any(LoadBalanceRequest.class));
  assertEquals(0, fakeClock.numPendingTasks(LOAD_REPORTING_TASK_FILTER));
}

/**
 * If a {@link SubchannelPicker} is being, or has been provided via {@link #reprocess}, the last
 * picker will be consulted.
 *
 * <p>Otherwise, if the delayed transport is not shutdown, then a {@link PendingStream} is
 * returned; if the transport is shutdown, then a {@link FailingClientStream} is returned.
 */
@Override
public final ClientStream newStream(
    MethodDescriptor<?, ?> method, Metadata headers, CallOptions callOptions) {
  try {
    SubchannelPicker picker;
    PickSubchannelArgs args = new PickSubchannelArgsImpl(method, headers, callOptions);
    long pickerVersion = -1;
    synchronized (lock) {
      if (shutdownStatus == null) {
        if (lastPicker == null) {
          return createPendingStream(args);
        }
        picker = lastPicker;
        pickerVersion = lastPickerVersion;
      } else {
        return new FailingClientStream(shutdownStatus);
      }
    }
    while (true) {
      PickResult pickResult = picker.pickSubchannel(args);
      ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
          callOptions.isWaitForReady());
      if (transport != null) {
        return transport.newStream(
            args.getMethodDescriptor(), args.getHeaders(), args.getCallOptions());
      }
      // This picker's conclusion is "buffer".  If there hasn't been a newer picker set (possible
      // race with reprocess()), we will buffer it.  Otherwise, will try with the new picker.
      synchronized (lock) {
        if (shutdownStatus != null) {
          return new FailingClientStream(shutdownStatus);
        }
        if (pickerVersion == lastPickerVersion) {
          return createPendingStream(args);
        }
        picker = lastPicker;
        pickerVersion = lastPickerVersion;
      }
    }
  } finally {
    channelExecutor.drain();
  }
}

private PendingStream(PickSubchannelArgs args) {
  this.args = args;
}

void setSubchannel(final InternalSubchannel subchannel) {
  log.log(Level.FINE, "[{0}] Created with [{1}]", new Object[] {this, subchannel});
  this.subchannel = subchannel;
  subchannelImpl = new AbstractSubchannel() {
      @Override
      public void shutdown() {
        subchannel.shutdown(Status.UNAVAILABLE.withDescription("OobChannel is shutdown"));
      }

      @Override
      ClientTransport obtainActiveTransport() {
        return subchannel.obtainActiveTransport();
      }

      @Override
      public void requestConnection() {
        subchannel.obtainActiveTransport();
      }

      @Override
      public EquivalentAddressGroup getAddresses() {
        return subchannel.getAddressGroup();
      }

      @Override
      public Attributes getAttributes() {
        return Attributes.EMPTY;
      }

      @Override
      public ListenableFuture<ChannelStats> getStats() {
        SettableFuture<ChannelStats> ret = SettableFuture.create();
        ChannelStats.Builder builder = new ChannelStats.Builder();
        subchannelCallsTracer.updateBuilder(builder);
        builder.setTarget(authority).setState(subchannel.getState());
        ret.set(builder.build());
        return ret;
      }
  };

  subchannelPicker = new SubchannelPicker() {
      final PickResult result = PickResult.withSubchannel(subchannelImpl);

      @Override
      public PickResult pickSubchannel(PickSubchannelArgs args) {
        return result;
      }
    };
  delayedTransport.reprocess(subchannelPicker);
}

@Override
public PickResult pickSubchannel(PickSubchannelArgs args) {
  return result;
}

@Test
public void callOptionsExecutor() {
  Metadata headers = new Metadata();
  ClientStream mockStream = mock(ClientStream.class);
  FakeClock callExecutor = new FakeClock();
  createChannel(new FakeNameResolverFactory(true), NO_INTERCEPTOR);

  // Start a call with a call executor
  CallOptions options =
      CallOptions.DEFAULT.withExecutor(callExecutor.getScheduledExecutorService());
  ClientCall<String, Integer> call = channel.newCall(method, options);
  call.start(mockCallListener, headers);

  // Make the transport available
  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  verify(mockTransportFactory, never()).newClientTransport(
      any(SocketAddress.class), any(String.class), any(String.class), any(ProxyParameters.class));
  subchannel.requestConnection();
  verify(mockTransportFactory).newClientTransport(
      any(SocketAddress.class), any(String.class), any(String.class), any(ProxyParameters.class));
  MockClientTransportInfo transportInfo = transports.poll();
  ConnectionClientTransport mockTransport = transportInfo.transport;
  ManagedClientTransport.Listener transportListener = transportInfo.listener;
  when(mockTransport.newStream(same(method), same(headers), any(CallOptions.class)))
      .thenReturn(mockStream);
  transportListener.transportReady();
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel));
  assertEquals(0, callExecutor.numPendingTasks());
  helper.updateBalancingState(READY, mockPicker);

  // Real streams are started in the call executor if they were previously buffered.
  assertEquals(1, callExecutor.runDueTasks());
  verify(mockTransport).newStream(same(method), same(headers), same(options));
  verify(mockStream).start(streamListenerCaptor.capture());

  // Call listener callbacks are also run in the call executor
  ClientStreamListener streamListener = streamListenerCaptor.getValue();
  Metadata trailers = new Metadata();
  assertEquals(0, callExecutor.numPendingTasks());
  streamListener.closed(Status.CANCELLED, trailers);
  verify(mockCallListener, never()).onClose(same(Status.CANCELLED), same(trailers));
  assertEquals(1, callExecutor.runDueTasks());
  verify(mockCallListener).onClose(same(Status.CANCELLED), same(trailers));


  transportListener.transportShutdown(Status.UNAVAILABLE);
  transportListener.transportTerminated();

  // Clean up as much as possible to allow the channel to terminate.
  subchannel.shutdown();
  timer.forwardNanos(
      TimeUnit.SECONDS.toNanos(ManagedChannelImpl.SUBCHANNEL_SHUTDOWN_DELAY_SECONDS));
}

/**
 * Verify that if the first resolved address points to a server that cannot be connected, the call
 * will end up with the second address which works.
 */
@Test
public void firstResolvedServerFailedToConnect() throws Exception {
  final SocketAddress goodAddress = new SocketAddress() {
      @Override public String toString() {
        return "goodAddress";
      }
    };
  final SocketAddress badAddress = new SocketAddress() {
      @Override public String toString() {
        return "badAddress";
      }
    };
  InOrder inOrder = inOrder(mockLoadBalancer);

  List<SocketAddress> resolvedAddrs = Arrays.asList(badAddress, goodAddress);
  FakeNameResolverFactory nameResolverFactory = new FakeNameResolverFactory(resolvedAddrs);
  createChannel(nameResolverFactory, NO_INTERCEPTOR);

  // Start the call
  ClientCall<String, Integer> call = channel.newCall(method, CallOptions.DEFAULT);
  Metadata headers = new Metadata();
  call.start(mockCallListener, headers);
  executor.runDueTasks();

  // Simulate name resolution results
  EquivalentAddressGroup addressGroup = new EquivalentAddressGroup(resolvedAddrs);
  inOrder.verify(mockLoadBalancer).handleResolvedAddressGroups(
      eq(Arrays.asList(addressGroup)), eq(Attributes.EMPTY));
  Subchannel subchannel = helper.createSubchannel(addressGroup, Attributes.EMPTY);
  when(mockPicker.pickSubchannel(any(PickSubchannelArgs.class)))
      .thenReturn(PickResult.withSubchannel(subchannel));
  subchannel.requestConnection();
  inOrder.verify(mockLoadBalancer).handleSubchannelState(
      same(subchannel), stateInfoCaptor.capture());
  assertEquals(CONNECTING, stateInfoCaptor.getValue().getState());

  // The channel will starts with the first address (badAddress)
  verify(mockTransportFactory)
      .newClientTransport(same(badAddress), any(String.class), any(String.class),
          any(ProxyParameters.class));
  verify(mockTransportFactory, times(0))
        .newClientTransport(same(goodAddress), any(String.class), any(String.class),
            any(ProxyParameters.class));

  MockClientTransportInfo badTransportInfo = transports.poll();
  // Which failed to connect
  badTransportInfo.listener.transportShutdown(Status.UNAVAILABLE);
  inOrder.verifyNoMoreInteractions();

  // The channel then try the second address (goodAddress)
  verify(mockTransportFactory)
        .newClientTransport(same(goodAddress), any(String.class), any(String.class),
            any(ProxyParameters.class));
  MockClientTransportInfo goodTransportInfo = transports.poll();
  when(goodTransportInfo.transport.newStream(
          any(MethodDescriptor.class), any(Metadata.class), any(CallOptions.class)))
      .thenReturn(mock(ClientStream.class));

  goodTransportInfo.listener.transportReady();
  inOrder.verify(mockLoadBalancer).handleSubchannelState(
      same(subchannel), stateInfoCaptor.capture());
  assertEquals(READY, stateInfoCaptor.getValue().getState());

  // A typical LoadBalancer will call this once the subchannel becomes READY
  helper.updateBalancingState(READY, mockPicker);
  // Delayed transport uses the app executor to create real streams.
  executor.runDueTasks();

  verify(goodTransportInfo.transport).newStream(same(method), same(headers),
      same(CallOptions.DEFAULT));
  // The bad transport was never used.
  verify(badTransportInfo.transport, times(0)).newStream(any(MethodDescriptor.class),
      any(Metadata.class), any(CallOptions.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();
}

/**
 * Returns a transport for a new call.
 *
 * @param args object containing call arguments.
 */
ClientTransport get(PickSubchannelArgs args);