Java 类io.grpc.ConnectivityStateInfo 实例源码

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

@Override
public void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
  if (!subchannels.containsValue(subchannel)) {
    return;
  }
  if (stateInfo.getState() == IDLE) {
    subchannel.requestConnection();
  }
  getSubchannelStateInfoRef(subchannel).set(stateInfo);
  updateBalancingState(getAggregatedState(), getAggregatedError());
}

/**
 * If all subchannels are TRANSIENT_FAILURE, return the Status associated with an arbitrary
 * subchannel otherwise, return null.
 */
@Nullable
private Status getAggregatedError() {
  Status status = null;
  for (Subchannel subchannel : getSubchannels()) {
    ConnectivityStateInfo stateInfo = getSubchannelStateInfoRef(subchannel).get();
    if (stateInfo.getState() != TRANSIENT_FAILURE) {
      return null;
    }
    status = stateInfo.getStatus();
  }
  return status;
}

@Override
public void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState) {
  if (delegate != null) {
    delegate.handleSubchannelState(subchannel, newState);
    return;
  }
  if (grpclbState != null) {
    grpclbState.handleSubchannelState(subchannel, newState);
  }
}

void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo newState) {
  if (newState.getState() == SHUTDOWN || !(subchannels.values().contains(subchannel))) {
    return;
  }
  if (newState.getState() == IDLE) {
    subchannel.requestConnection();
  }
  subchannel.getAttributes().get(STATE_INFO).set(newState);
  maybeStartFallbackTimer();
  maybeUpdatePicker();
}

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

private void deliverSubchannelState(
    final Subchannel subchannel, final ConnectivityStateInfo newState) {
  channelExecutor.execute(new Runnable() {
      @Override
      public void run() {
        balancer.handleSubchannelState(subchannel, newState);
      }
    });
}

@GuardedBy("lock")
private void gotoState(final ConnectivityStateInfo newState) {
  if (state.getState() != newState.getState()) {
    Preconditions.checkState(state.getState() != SHUTDOWN,
        "Cannot transition out of SHUTDOWN to " + newState);
    state = newState;
    channelExecutor.executeLater(new Runnable() {
        @Override
        public void run() {
          callback.onStateChange(InternalSubchannel.this, newState);
        }
      });
  }
}

@Override
public void handleSubchannelState(Subchannel subchannel, ConnectivityStateInfo stateInfo) {
  if (subchannels.get(subchannel.getAddresses()) != subchannel) {
    return;
  }
  if (stateInfo.getState() == IDLE) {
    subchannel.requestConnection();
  }
  getSubchannelStateInfoRef(subchannel).value = stateInfo;
  updateBalancingState(getAggregatedState(), getAggregatedError());
}

/**
 * If all subchannels are TRANSIENT_FAILURE, return the Status associated with an arbitrary
 * subchannel otherwise, return null.
 */
@Nullable
private Status getAggregatedError() {
  Status status = null;
  for (Subchannel subchannel : getSubchannels()) {
    ConnectivityStateInfo stateInfo = getSubchannelStateInfoRef(subchannel).value;
    if (stateInfo.getState() != TRANSIENT_FAILURE) {
      return null;
    }
    status = stateInfo.getStatus();
  }
  return status;
}

@Test
public void pickAfterStateChange() throws Exception {
  InOrder inOrder = inOrder(mockHelper);
  loadBalancer.handleResolvedAddressGroups(servers, Attributes.EMPTY);
  Subchannel subchannel = loadBalancer.getSubchannels().iterator().next();
  Ref<ConnectivityStateInfo> subchannelStateInfo = subchannel.getAttributes().get(
      STATE_INFO);

  inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), isA(Picker.class));
  assertThat(subchannelStateInfo.value).isEqualTo(ConnectivityStateInfo.forNonError(IDLE));

  loadBalancer.handleSubchannelState(subchannel,
      ConnectivityStateInfo.forNonError(READY));
  inOrder.verify(mockHelper).updateBalancingState(eq(READY), pickerCaptor.capture());
  assertNull(pickerCaptor.getValue().getStatus());
  assertThat(subchannelStateInfo.value).isEqualTo(
      ConnectivityStateInfo.forNonError(READY));

  Status error = Status.UNKNOWN.withDescription("¯\\_(ツ)_//¯");
  loadBalancer.handleSubchannelState(subchannel,
      ConnectivityStateInfo.forTransientFailure(error));
  assertThat(subchannelStateInfo.value).isEqualTo(
      ConnectivityStateInfo.forTransientFailure(error));
  inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
  assertNull(pickerCaptor.getValue().getStatus());

  loadBalancer.handleSubchannelState(subchannel,
      ConnectivityStateInfo.forNonError(IDLE));
  inOrder.verify(mockHelper).updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
  assertNull(pickerCaptor.getValue().getStatus());
  assertThat(subchannelStateInfo.value).isEqualTo(
      ConnectivityStateInfo.forNonError(IDLE));

  verify(subchannel, times(2)).requestConnection();
  verify(mockHelper, times(3)).createSubchannel(any(EquivalentAddressGroup.class),
      any(Attributes.class));
  verifyNoMoreInteractions(mockHelper);
}

private static AtomicReference<ConnectivityStateInfo> getSubchannelStateInfoRef(
    Subchannel subchannel) {
  return checkNotNull(subchannel.getAttributes().get(STATE_INFO), "STATE_INFO");
}

private void handleInternalSubchannelState(ConnectivityStateInfo newState) {
  if (newState.getState() == TRANSIENT_FAILURE || newState.getState() == IDLE) {
    nr.refresh();
  }
}

/**
 * Only called after all addresses attempted and failed (TRANSIENT_FAILURE).
 * @param status the causal status when the channel begins transition to
 *     TRANSIENT_FAILURE.
 */
@GuardedBy("lock")
private void scheduleBackoff(final Status status) {
  class EndOfCurrentBackoff implements Runnable {
    @Override
    public void run() {
      try {
        synchronized (lock) {
          reconnectTask = null;
          if (reconnectCanceled) {
            // Even though cancelReconnectTask() will cancel this task, the task may have already
            // started when it's being canceled.
            return;
          }
          gotoNonErrorState(CONNECTING);
          startNewTransport();
        }
      } catch (Throwable t) {
        log.log(Level.WARNING, "Exception handling end of backoff", t);
      } finally {
        channelExecutor.drain();
      }
    }
  }

  gotoState(ConnectivityStateInfo.forTransientFailure(status));
  if (reconnectPolicy == null) {
    reconnectPolicy = backoffPolicyProvider.get();
  }
  long delayNanos =
      reconnectPolicy.nextBackoffNanos() - connectingTimer.elapsed(TimeUnit.NANOSECONDS);
  if (log.isLoggable(Level.FINE)) {
    log.log(Level.FINE, "[{0}] Scheduling backoff for {1} ns", new Object[]{logId, delayNanos});
  }
  Preconditions.checkState(reconnectTask == null, "previous reconnectTask is not done");
  reconnectCanceled = false;
  reconnectTask = scheduledExecutor.schedule(
      new LogExceptionRunnable(new EndOfCurrentBackoff()),
      delayNanos,
      TimeUnit.NANOSECONDS);
}

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

private static Ref<ConnectivityStateInfo> getSubchannelStateInfoRef(
    Subchannel subchannel) {
  return checkNotNull(subchannel.getAttributes().get(STATE_INFO), "STATE_INFO");
}

@Override
protected void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) {
  assertSame(internalSubchannel, is);
  callbackInvokes.add("onStateChange:" + newState);
}

@Test
public void pickAfterResolvedUpdatedHosts() throws Exception {
  Subchannel removedSubchannel = mock(Subchannel.class);
  Subchannel oldSubchannel = mock(Subchannel.class);
  Subchannel newSubchannel = mock(Subchannel.class);

  for (Subchannel subchannel : Lists.newArrayList(removedSubchannel, oldSubchannel,
      newSubchannel)) {
    when(subchannel.getAttributes()).thenReturn(Attributes.newBuilder().set(STATE_INFO,
        new Ref<ConnectivityStateInfo>(
            ConnectivityStateInfo.forNonError(READY))).build());
  }

  FakeSocketAddress removedAddr = new FakeSocketAddress("removed");
  FakeSocketAddress oldAddr = new FakeSocketAddress("old");
  FakeSocketAddress newAddr = new FakeSocketAddress("new");

  final Map<EquivalentAddressGroup, Subchannel> subchannels2 = Maps.newHashMap();
  subchannels2.put(new EquivalentAddressGroup(removedAddr), removedSubchannel);
  subchannels2.put(new EquivalentAddressGroup(oldAddr), oldSubchannel);

  List<EquivalentAddressGroup> currentServers =
      Lists.newArrayList(
          new EquivalentAddressGroup(removedAddr),
          new EquivalentAddressGroup(oldAddr));

  doAnswer(new Answer<Subchannel>() {
    @Override
    public Subchannel answer(InvocationOnMock invocation) throws Throwable {
      Object[] args = invocation.getArguments();
      return subchannels2.get(args[0]);
    }
  }).when(mockHelper).createSubchannel(any(EquivalentAddressGroup.class), any(Attributes.class));

  loadBalancer.handleResolvedAddressGroups(currentServers, affinity);

  InOrder inOrder = inOrder(mockHelper);

  inOrder.verify(mockHelper).updateBalancingState(eq(READY), pickerCaptor.capture());
  Picker picker = pickerCaptor.getValue();
  assertNull(picker.getStatus());
  assertThat(picker.getList()).containsExactly(removedSubchannel, oldSubchannel);

  verify(removedSubchannel, times(1)).requestConnection();
  verify(oldSubchannel, times(1)).requestConnection();

  assertThat(loadBalancer.getSubchannels()).containsExactly(removedSubchannel,
      oldSubchannel);

  subchannels2.clear();
  subchannels2.put(new EquivalentAddressGroup(oldAddr), oldSubchannel);
  subchannels2.put(new EquivalentAddressGroup(newAddr), newSubchannel);

  List<EquivalentAddressGroup> latestServers =
      Lists.newArrayList(
          new EquivalentAddressGroup(oldAddr),
          new EquivalentAddressGroup(newAddr));

  loadBalancer.handleResolvedAddressGroups(latestServers, affinity);

  verify(newSubchannel, times(1)).requestConnection();
  verify(removedSubchannel, times(1)).shutdown();

  assertThat(loadBalancer.getSubchannels()).containsExactly(oldSubchannel,
      newSubchannel);

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

  picker = pickerCaptor.getValue();
  assertNull(picker.getStatus());
  assertThat(picker.getList()).containsExactly(oldSubchannel, newSubchannel);

  verifyNoMoreInteractions(mockHelper);
}

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

/**
 * Called when the subchannel's connectivity state has changed.
 */
@ForOverride
void onStateChange(InternalSubchannel is, ConnectivityStateInfo newState) { }