Java 类io.grpc.ConnectivityState 实例源码

private ConnectivityState getAggregatedState() {
  Set<ConnectivityState> states = EnumSet.noneOf(ConnectivityState.class);
  for (Subchannel subchannel : getSubchannels()) {
    states.add(getSubchannelStateInfoRef(subchannel).get().getState());
  }
  if (states.contains(READY)) {
    return READY;
  }
  if (states.contains(CONNECTING)) {
    return CONNECTING;
  }
  if (states.contains(IDLE)) {
    return CONNECTING;
  }
  return TRANSIENT_FAILURE;
}

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

@Override
public void updateBalancingState(
    final ConnectivityState newState, final SubchannelPicker newPicker) {
  checkNotNull(newState, "newState");
  checkNotNull(newPicker, "newPicker");

  runSerialized(
      new Runnable() {
        @Override
        public void run() {
          if (LbHelperImpl.this != lbHelper) {
            return;
          }
          subchannelPicker = newPicker;
          delayedTransport.reprocess(newPicker);
          // It's not appropriate to report SHUTDOWN state from lb.
          // Ignore the case of newState == SHUTDOWN for now.
          if (newState != SHUTDOWN) {
            channelStateManager.gotoState(newState);
          }
        }
      });
}

private ConnectivityState getAggregatedState() {
  Set<ConnectivityState> states = EnumSet.noneOf(ConnectivityState.class);
  for (Subchannel subchannel : getSubchannels()) {
    states.add(getSubchannelStateInfoRef(subchannel).value.getState());
  }
  if (states.contains(READY)) {
    return READY;
  }
  if (states.contains(CONNECTING)) {
    return CONNECTING;
  }
  if (states.contains(IDLE)) {
    // This subchannel IDLE is not because of channel IDLE_TIMEOUT, in which case LB is already
    // shutdown.
    // RRLB will request connection immediately on subchannel IDLE.
    return CONNECTING;
  }
  return TRANSIENT_FAILURE;
}

@Test
public void proxyTest() {
  final SocketAddress addr1 = mock(SocketAddress.class);
  final ProxyParameters proxy = new ProxyParameters(
      InetSocketAddress.createUnresolved("proxy.example.com", 1000), "username", "password");
  ProxyDetector proxyDetector = new ProxyDetector() {
    @Nullable
    @Override
    public ProxyParameters proxyFor(SocketAddress targetServerAddress) {
      if (targetServerAddress == addr1) {
        return proxy;
      } else {
        return null;
      }
    }
  };
  createInternalSubChannelWithProxy(proxyDetector, addr1);
  assertEquals(ConnectivityState.IDLE, internalSubchannel.getState());
  assertNoCallbackInvoke();
  assertNull(internalSubchannel.obtainActiveTransport());
  assertExactCallbackInvokes("onStateChange:CONNECTING");
  assertEquals(ConnectivityState.CONNECTING, internalSubchannel.getState());
  verify(mockTransportFactory).newClientTransport(
      eq(addr1), eq(AUTHORITY), eq(USER_AGENT), eq(proxy));
}

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

/**
 * Make and use a picker out of the current lists and the states of subchannels if they have
 * changed since the last picker created.
 */
private void maybeUpdatePicker() {
  List<RoundRobinEntry> pickList = new ArrayList<RoundRobinEntry>(backendList.size());
  Status error = null;
  boolean hasIdle = false;
  for (BackendEntry entry : backendList) {
    Subchannel subchannel = entry.result.getSubchannel();
    Attributes attrs = subchannel.getAttributes();
    ConnectivityStateInfo stateInfo = attrs.get(STATE_INFO).get();
    if (stateInfo.getState() == READY) {
      pickList.add(entry);
    } else if (stateInfo.getState() == TRANSIENT_FAILURE) {
      error = stateInfo.getStatus();
    } else if (stateInfo.getState() == IDLE) {
      hasIdle = true;
    }
  }
  ConnectivityState state;
  if (pickList.isEmpty()) {
    if (error != null && !hasIdle) {
      logger.log(Level.FINE, "[{0}] No ready Subchannel. Using error: {1}",
          new Object[] {logId, error});
      pickList.add(new ErrorEntry(error));
      state = TRANSIENT_FAILURE;
    } else {
      logger.log(Level.FINE, "[{0}] No ready Subchannel and still connecting", logId);
      pickList.add(BUFFER_ENTRY);
      state = CONNECTING;
    }
  } else {
    logger.log(
        Level.FINE, "[{0}] Using drop list {1} and pick list {2}",
        new Object[] {logId, dropList, pickList});
    state = READY;
  }
  maybeUpdatePicker(state, new RoundRobinPicker(dropList, pickList));
}

/**
 * Update the given picker to the helper if it's different from the current one.
 */
private void maybeUpdatePicker(ConnectivityState state, RoundRobinPicker picker) {
  // Discard the new picker if we are sure it won't make any difference, in order to save
  // re-processing pending streams, and avoid unnecessary resetting of the pointer in
  // RoundRobinPicker.
  if (picker.dropList.equals(currentPicker.dropList)
      && picker.pickList.equals(currentPicker.pickList)) {
    return;
  }
  // No need to skip ErrorPicker. If the current picker is ErrorPicker, there won't be any pending
  // stream thus no time is wasted in re-process.
  currentPicker = picker;
  helper.updateBalancingState(state, picker);
}

@Override
public ConnectivityState getState(boolean requestConnection) {
  ConnectivityState savedChannelState = channelStateManager.getState();
  if (requestConnection && savedChannelState == IDLE) {
    channelExecutor.executeLater(
        new Runnable() {
          @Override
          public void run() {
            exitIdleMode();
          }
        }).drain();
  }
  return savedChannelState;
}

@Override
public void notifyWhenStateChanged(final ConnectivityState source, final Runnable callback) {
  channelExecutor.executeLater(
      new Runnable() {
        @Override
        public void run() {
          channelStateManager.notifyWhenStateChanged(callback, executor, source);
        }
      }).drain();
}

@VisibleForTesting
ConnectivityState getState() {
  try {
    synchronized (lock) {
      return state.getState();
    }
  } finally {
    channelExecutor.drain();
  }
}

/**
 * Creates an instance.
 */
public ChannelStats(
    String target,
    ConnectivityState state,
    long callsStarted,
    long callsSucceeded,
    long callsFailed,
    long lastCallStartedMillis) {
  this.target = target;
  this.state = state;
  this.callsStarted = callsStarted;
  this.callsSucceeded = callsSucceeded;
  this.callsFailed = callsFailed;
  this.lastCallStartedMillis = lastCallStartedMillis;
}

/**
 * Adds a listener for state change event.
 *
 * <p>The {@code executor} must be one that can run RPC call listeners.
 */
void notifyWhenStateChanged(Runnable callback, Executor executor, ConnectivityState source) {
  checkNotNull(callback, "callback");
  checkNotNull(executor, "executor");
  checkNotNull(source, "source");

  Listener stateChangeListener = new Listener(callback, executor);
  if (state != source) {
    stateChangeListener.runInExecutor();
  } else {
    listeners.add(stateChangeListener);
  }
}

private void gotoNullableState(@Nullable ConnectivityState newState) {
  if (state != newState && state != ConnectivityState.SHUTDOWN) {
    state = newState;
    if (listeners.isEmpty()) {
      return;
    }
    // Swap out callback list before calling them, because a callback may register new callbacks,
    // if run in direct executor, can cause ConcurrentModificationException.
    ArrayList<Listener> savedListeners = listeners;
    listeners = new ArrayList<Listener>();
    for (Listener listener : savedListeners) {
      listener.runInExecutor();
    }
  }
}

/**
 * Gets the current connectivity state of the channel. This method is threadsafe.
 */
ConnectivityState getState() {
  ConnectivityState stateCopy = state;
  if (stateCopy == null) {
    throw new UnsupportedOperationException("Channel state API is not implemented");
  }
  return stateCopy;
}

/**
 * Updates picker with the list of active subchannels (state == READY).
 */
private void updateBalancingState(ConnectivityState state, Status error) {
  List<Subchannel> activeList = filterNonFailingSubchannels(getSubchannels());
  helper.updateBalancingState(state, new GrpcRoutePicker(activeList, error, attributes));
}

@Test
public void resetGrpclbWhenSwitchingAwayFromGrpclb() {
  InOrder inOrder = inOrder(helper);
  List<EquivalentAddressGroup> grpclbResolutionList = createResolvedServerAddresses(true);
  Attributes grpclbResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.GRPCLB).build();
  deliverResolvedAddresses(grpclbResolutionList, grpclbResolutionAttrs);

  assertSame(LbPolicy.GRPCLB, balancer.getLbPolicy());
  assertNull(balancer.getDelegate());
  verify(helper).createOobChannel(addrsEq(grpclbResolutionList.get(0)), eq(lbAuthority(0)));
  assertEquals(1, fakeOobChannels.size());
  ManagedChannel oobChannel = fakeOobChannels.poll();
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());
  StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue();
  assertEquals(1, lbRequestObservers.size());
  StreamObserver<LoadBalanceRequest> lbRequestObserver = lbRequestObservers.poll();
  verify(lbRequestObserver).onNext(
      eq(LoadBalanceRequest.newBuilder().setInitialRequest(
              InitialLoadBalanceRequest.newBuilder().setName(SERVICE_AUTHORITY).build())
          .build()));

  // Simulate receiving LB response
  List<ServerEntry> backends = Arrays.asList(new ServerEntry("127.0.0.1", 2000, "token0001"));
  inOrder.verify(helper, never())
      .updateBalancingState(any(ConnectivityState.class), any(SubchannelPicker.class));
  lbResponseObserver.onNext(buildInitialResponse());
  lbResponseObserver.onNext(buildLbResponse(backends));

  inOrder.verify(helper).createSubchannel(
      eq(new EquivalentAddressGroup(backends.get(0).addr)), any(Attributes.class));
  assertEquals(1, mockSubchannels.size());
  Subchannel subchannel = mockSubchannels.poll();
  verify(subchannel).requestConnection();

  // Switch to round-robin. GRPCLB streams and connections should be closed. 
  List<EquivalentAddressGroup> roundRobinResolutionList =
      createResolvedServerAddresses(false, false, false);
  Attributes roundRobinResolutionAttrs = Attributes.newBuilder()
      .set(GrpclbConstants.ATTR_LB_POLICY, LbPolicy.ROUND_ROBIN).build();
  verify(lbRequestObserver, never()).onCompleted();
  verify(subchannel, never()).shutdown();
  assertFalse(oobChannel.isShutdown());
  deliverResolvedAddresses(roundRobinResolutionList, roundRobinResolutionAttrs);

  verify(lbRequestObserver).onCompleted();
  verify(subchannel).shutdown();
  assertTrue(oobChannel.isShutdown());
  assertTrue(oobChannel.isTerminated());
  assertSame(LbPolicy.ROUND_ROBIN, balancer.getLbPolicy());
  assertSame(roundRobinBalancer, balancer.getDelegate());
  assertNull(balancer.getGrpclbState());
}

@GuardedBy("lock")
private void gotoNonErrorState(ConnectivityState newState) {
  gotoState(ConnectivityStateInfo.forNonError(newState));
}

public Builder setState(ConnectivityState state) {
  this.state = state;
  return this;
}

/**
 * Connectivity state is changed to the specified value. Will trigger some notifications that have
 * been registered earlier by {@link ManagedChannel#notifyWhenStateChanged}.
 */
void gotoState(@Nonnull ConnectivityState newState) {
  checkNotNull(newState, "newState");
  checkState(!isDisabled(), "ConnectivityStateManager is already disabled");
  gotoNullableState(newState);
}

/**
 * Updates picker with the list of active subchannels (state == READY).
 */
private void updateBalancingState(ConnectivityState state, Status error) {
  List<Subchannel> activeList = filterNonFailingSubchannels(getSubchannels());
  helper.updateBalancingState(state, new Picker(activeList, error));
}

@Test
public void subchannelStateIsolation() throws Exception {
  Iterator<Subchannel> subchannelIterator = subchannels.values().iterator();
  Subchannel sc1 = subchannelIterator.next();
  Subchannel sc2 = subchannelIterator.next();
  Subchannel sc3 = subchannelIterator.next();

  loadBalancer.handleResolvedAddressGroups(servers, Attributes.EMPTY);
  verify(sc1, times(1)).requestConnection();
  verify(sc2, times(1)).requestConnection();
  verify(sc3, times(1)).requestConnection();

  loadBalancer.handleSubchannelState(sc1, ConnectivityStateInfo.forNonError(READY));
  loadBalancer.handleSubchannelState(sc2, ConnectivityStateInfo.forNonError(READY));
  loadBalancer.handleSubchannelState(sc3, ConnectivityStateInfo.forNonError(READY));
  loadBalancer.handleSubchannelState(sc2, ConnectivityStateInfo.forNonError(IDLE));
  loadBalancer
      .handleSubchannelState(sc3, ConnectivityStateInfo.forTransientFailure(Status.UNAVAILABLE));

  verify(mockHelper, times(6))
      .updateBalancingState(stateCaptor.capture(), pickerCaptor.capture());
  Iterator<ConnectivityState> stateIterator = stateCaptor.getAllValues().iterator();
  Iterator<Picker> pickers = pickerCaptor.getAllValues().iterator();
  // The picker is incrementally updated as subchannels become READY
  assertEquals(CONNECTING, stateIterator.next());
  assertThat(pickers.next().getList()).isEmpty();
  assertEquals(READY, stateIterator.next());
  assertThat(pickers.next().getList()).containsExactly(sc1);
  assertEquals(READY, stateIterator.next());
  assertThat(pickers.next().getList()).containsExactly(sc1, sc2);
  assertEquals(READY, stateIterator.next());
  assertThat(pickers.next().getList()).containsExactly(sc1, sc2, sc3);
  // The IDLE subchannel is dropped from the picker, but a reconnection is requested
  assertEquals(READY, stateIterator.next());
  assertThat(pickers.next().getList()).containsExactly(sc1, sc3);
  verify(sc2, times(2)).requestConnection();
  // The failing subchannel is dropped from the picker, with no requested reconnect
  assertEquals(READY, stateIterator.next());
  assertThat(pickers.next().getList()).containsExactly(sc1);
  verify(sc3, times(1)).requestConnection();
  assertThat(stateIterator.hasNext()).isFalse();
  assertThat(pickers.hasNext()).isFalse();
}