io.grpc.Status#UNAVAILABLE源码实例Demo

private void updatePicker(
    @Nullable ConnectivityState state,  List<WeightedChildPicker> childPickers) {
  SubchannelPicker picker;
  if (childPickers.isEmpty()) {
    if (state == TRANSIENT_FAILURE) {
      picker = new ErrorPicker(Status.UNAVAILABLE); // TODO: more details in status
    } else {
      picker = XdsSubchannelPickers.BUFFER_PICKER;
    }
  } else {
    picker = new WeightedRandomPicker(childPickers);
  }

  if (!dropOverloads.isEmpty()) {
    picker = new DroppablePicker(dropOverloads, picker, random, loadStatsStore);
  }

  if (state != null) {
    helper.updateBalancingState(state, picker);
  }
}
 

@Test
public void testRetryOnUnavailable() throws IOException {
  String uniqueName = InProcessServerBuilder.generateName();
  ExecutionImpl service = new ExecutionImpl(Status.UNAVAILABLE, 0);
  InProcessServerBuilder.forName(uniqueName).addService(service).build().start();
  CallCounter beforeRetry = new CallCounter();
  ManagedChannel channel =
      InProcessChannelBuilder.forName(uniqueName)
          .intercept(
              new RetryClientInterceptor(
                  RetryPolicy.builder().setMaxRetries(2).setBeforeRetry(beforeRetry).build()))
          .build();
  ExecutionBlockingStub stub = ExecutionGrpc.newBlockingStub(channel);
  try {
    stub.execute(ExecuteRequest.newBuilder().build()).forEachRemaining(resp -> {});
    Assert.fail("Final retry should cause an exception");
  } catch (StatusRuntimeException ex) {
    Assert.assertEquals(Status.Code.UNAVAILABLE, ex.getStatus().getCode());
  }

  Assert.assertEquals(3, service.calls);
  Assert.assertEquals(2, beforeRetry.calls);
}
 

@Test
public void getTransportFromPickResult_errorPickResult_waitForReady() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withError(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, true);

  assertNull(transport);
}
 

@Test
public void getTransportFromPickResult_errorPickResult_failFast() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withError(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, false);

  assertNotNull(transport);

  ClientStream stream = transport
      .newStream(TestMethodDescriptors.voidMethod(), new Metadata(), CallOptions.DEFAULT);
  ClientStreamListener listener = mock(ClientStreamListener.class);
  stream.start(listener);

  verify(listener).closed(eq(status), eq(RpcProgress.PROCESSED), any(Metadata.class));
}
 

@Test
public void getTransportFromPickResult_dropPickResult_waitForReady() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withDrop(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, true);

  assertNotNull(transport);

  ClientStream stream = transport
      .newStream(TestMethodDescriptors.voidMethod(), new Metadata(), CallOptions.DEFAULT);
  ClientStreamListener listener = mock(ClientStreamListener.class);
  stream.start(listener);

  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void getTransportFromPickResult_dropPickResult_failFast() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withDrop(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, false);

  assertNotNull(transport);

  ClientStream stream = transport
      .newStream(TestMethodDescriptors.voidMethod(), new Metadata(), CallOptions.DEFAULT);
  ClientStreamListener listener = mock(ClientStreamListener.class);
  stream.start(listener);

  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void processedRpcProgressPopulatedToListener() {
  ClientStreamListener listener = mock(ClientStreamListener.class);
  Status status = Status.UNAVAILABLE;

  ClientStream stream = new FailingClientStream(status);
  stream.start(listener);
  verify(listener).closed(eq(status), eq(RpcProgress.PROCESSED), any(Metadata.class));
}
 

@Test
public void droppedRpcProgressPopulatedToListener() {
  ClientStreamListener listener = mock(ClientStreamListener.class);
  Status status = Status.UNAVAILABLE;

  ClientStream stream = new FailingClientStream(status, RpcProgress.DROPPED);
  stream.start(listener);
  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void testRetryOnPartialResponse() throws IOException {
  String uniqueName = InProcessServerBuilder.generateName();
  ExecutionImpl service = new ExecutionImpl(Status.UNAVAILABLE, 1);
  InProcessServerBuilder.forName(uniqueName).addService(service).build().start();
  CallCounter beforeRetry = new CallCounter();
  ManagedChannel channel =
      InProcessChannelBuilder.forName(uniqueName)
          .intercept(
              new RetryClientInterceptor(
                  RetryPolicy.builder()
                      .setMaxRetries(2)
                      .setBeforeRetry(beforeRetry)
                      .setRestartAllStreamingCalls(true)
                      .build()))
          .build();
  ExecutionBlockingStub stub = ExecutionGrpc.newBlockingStub(channel);
  try {
    stub.execute(ExecuteRequest.newBuilder().build()).forEachRemaining(resp -> {});
    Assert.fail("Final retry should cause an exception");
  } catch (StatusRuntimeException ex) {
    Assert.assertEquals(Status.Code.UNAVAILABLE, ex.getStatus().getCode());
  }

  Assert.assertEquals(3, service.calls);
  Assert.assertEquals(2, beforeRetry.calls);
}
 

private Status toGrpcStatus(Throwable original) {
    Throwable cause = unwrap(original);
    if (cause instanceof SocketException) {
        return Status.UNAVAILABLE;
    } else if (cause instanceof TimeoutException) {
        return Status.DEADLINE_EXCEEDED;
    }
    for (Function<Throwable, Optional<Status>> mapper : serviceExceptionMappers) {
        Status status = mapper.apply(cause).orElse(null);
        if (status != null) {
            return status;
        }
    }
    return Status.INTERNAL;
}
 

private void updateOverallBalancingState() {
  List<WeightedChildPicker> childPickers = new ArrayList<>();

  ConnectivityState overallState = null;
  for (String name : targets.keySet()) {
    ChildHelper childHelper = childHelpers.get(name);
    ConnectivityState childState = childHelper.currentState;
    overallState = aggregateState(overallState, childState);
    if (READY == childState) {
      int weight = targets.get(name).weight;
      childPickers.add(new WeightedChildPicker(weight, childHelper.currentPicker));
    }
  }

  SubchannelPicker picker;
  if (childPickers.isEmpty()) {
    if (overallState == TRANSIENT_FAILURE) {
      picker = new ErrorPicker(Status.UNAVAILABLE); // TODO: more details in status
    } else {
      picker = XdsSubchannelPickers.BUFFER_PICKER;
    }
  } else {
    picker = new WeightedRandomPicker(childPickers);
  }

  if (overallState != null) {
    helper.updateBalancingState(overallState, picker);
  }
}
 

@Test
public void getTransportFromPickResult_errorPickResult_waitForReady() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withError(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, true);

  assertNull(transport);
}
 

@Test
public void getTransportFromPickResult_dropPickResult_waitForReady() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withDrop(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, true);

  assertNotNull(transport);

  ClientStream stream = transport
      .newStream(TestMethodDescriptors.voidMethod(), new Metadata(), CallOptions.DEFAULT);
  ClientStreamListener listener = mock(ClientStreamListener.class);
  stream.start(listener);

  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void getTransportFromPickResult_dropPickResult_failFast() {
  Status status = Status.UNAVAILABLE;
  PickResult pickResult = PickResult.withDrop(status);
  ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult, false);

  assertNotNull(transport);

  ClientStream stream = transport
      .newStream(TestMethodDescriptors.voidMethod(), new Metadata(), CallOptions.DEFAULT);
  ClientStreamListener listener = mock(ClientStreamListener.class);
  stream.start(listener);

  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void processedRpcProgressPopulatedToListener() {
  ClientStreamListener listener = mock(ClientStreamListener.class);
  Status status = Status.UNAVAILABLE;

  ClientStream stream = new FailingClientStream(status);
  stream.start(listener);
  verify(listener).closed(eq(status), eq(RpcProgress.PROCESSED), any(Metadata.class));
}
 

@Test
public void droppedRpcProgressPopulatedToListener() {
  ClientStreamListener listener = mock(ClientStreamListener.class);
  Status status = Status.UNAVAILABLE;

  ClientStream stream = new FailingClientStream(status, RpcProgress.DROPPED);
  stream.start(listener);
  verify(listener).closed(eq(status), eq(RpcProgress.DROPPED), any(Metadata.class));
}
 

@Test
public void statusRuntimeExceptionTransmitter() {
  final Status expectedStatus = Status.UNAVAILABLE;
  final Metadata expectedMetadata = new Metadata();
  FakeServerCall<Void, Void> call =
      new FakeServerCall<Void, Void>(expectedStatus, expectedMetadata);
  final StatusRuntimeException exception =
      new StatusRuntimeException(expectedStatus, expectedMetadata);
  listener = new VoidCallListener() {
    @Override
    public void onMessage(Void message) {
      throw exception;
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }

    @Override
    public void onCancel() {
      throw exception;
    }

    @Override
    public void onComplete() {
      throw exception;
    }

    @Override
    public void onReady() {
      throw exception;
    }
  };

  ServerServiceDefinition intercepted = ServerInterceptors.intercept(
      serviceDefinition,
      Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
  // The interceptor should have handled the error by directly closing the ServerCall
  // and the exception should not propagate to the method's caller
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onMessage(null);
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onCancel();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onComplete();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onHalfClose();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onReady();
  assertEquals(5, call.numCloses);
}
 

@Test
public void statusRuntimeExceptionTransmitterIgnoresClosedCalls() {
  final Status expectedStatus = Status.UNAVAILABLE;
  final Status unexpectedStatus = Status.CANCELLED;
  final Metadata expectedMetadata = new Metadata();

  FakeServerCall<Void, Void> call =
      new FakeServerCall<Void, Void>(expectedStatus, expectedMetadata);
  final StatusRuntimeException exception =
      new StatusRuntimeException(expectedStatus, expectedMetadata);

  listener = new VoidCallListener() {
    @Override
    public void onMessage(Void message) {
      throw exception;
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }
  };

  ServerServiceDefinition intercepted = ServerInterceptors.intercept(
      serviceDefinition,
      Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
  ServerCall.Listener<Void> callDoubleSreListener =
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
  callDoubleSreListener.onMessage(null); // the only close with our exception
  callDoubleSreListener.onHalfClose(); // should not trigger a close

  // this listener closes the call when it is initialized with startCall
  listener = new VoidCallListener() {
    @Override
    public void onCall(ServerCall<Void, Void> call, Metadata headers) {
      call.close(unexpectedStatus, headers);
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }
  };

  ServerCall.Listener<Void> callClosedListener =
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
  // call is already closed, does not match exception
  callClosedListener.onHalfClose(); // should not trigger a close
  assertEquals(1, call.numCloses);
}
 

@Test
public void statusRuntimeExceptionTransmitterIgnoresClosedCalls() {
  final Status expectedStatus = Status.UNAVAILABLE;
  final Status unexpectedStatus = Status.CANCELLED;
  final Metadata expectedMetadata = new Metadata();

  FakeServerCall<Void, Void> call =
      new FakeServerCall<>(expectedStatus, expectedMetadata);
  final StatusRuntimeException exception =
      new StatusRuntimeException(expectedStatus, expectedMetadata);

  listener = new VoidCallListener() {
    @Override
    public void onMessage(Void message) {
      throw exception;
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }
  };

  ServerServiceDefinition intercepted = ServerInterceptors.intercept(
      serviceDefinition,
      Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
  ServerCall.Listener<Void> callDoubleSreListener =
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
  callDoubleSreListener.onMessage(null); // the only close with our exception
  callDoubleSreListener.onHalfClose(); // should not trigger a close

  // this listener closes the call when it is initialized with startCall
  listener = new VoidCallListener() {
    @Override
    public void onCall(ServerCall<Void, Void> call, Metadata headers) {
      call.close(unexpectedStatus, headers);
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }
  };

  ServerCall.Listener<Void> callClosedListener =
      getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers);
  // call is already closed, does not match exception
  callClosedListener.onHalfClose(); // should not trigger a close
  assertEquals(1, call.numCloses);
}
 

@Test
public void statusRuntimeExceptionTransmitter() {
  final Status expectedStatus = Status.UNAVAILABLE;
  final Metadata expectedMetadata = new Metadata();
  FakeServerCall<Void, Void> call =
      new FakeServerCall<>(expectedStatus, expectedMetadata);
  final StatusRuntimeException exception =
      new StatusRuntimeException(expectedStatus, expectedMetadata);
  listener = new VoidCallListener() {
    @Override
    public void onMessage(Void message) {
      throw exception;
    }

    @Override
    public void onHalfClose() {
      throw exception;
    }

    @Override
    public void onCancel() {
      throw exception;
    }

    @Override
    public void onComplete() {
      throw exception;
    }

    @Override
    public void onReady() {
      throw exception;
    }
  };

  ServerServiceDefinition intercepted = ServerInterceptors.intercept(
      serviceDefinition,
      Arrays.asList(TransmitStatusRuntimeExceptionInterceptor.instance()));
  // The interceptor should have handled the error by directly closing the ServerCall
  // and the exception should not propagate to the method's caller
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onMessage(null);
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onCancel();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onComplete();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onHalfClose();
  getSoleMethod(intercepted).getServerCallHandler().startCall(call, headers).onReady();
  assertEquals(5, call.numCloses);
}