io.grpc.stub.StreamObserver#onNext ( )源码实例Demo

@Override
public void generic(Request request, StreamObserver<Response> responseObserver) {

    SofaRequest sofaRequest = TracingContextKey.getKeySofaRequest().get(Context.current());

    String methodName = sofaRequest.getMethodName();
    Class[] argTypes = getArgTypes(request);
    try {
        Serializer serializer = SerializerFactory.getSerializer(request.getSerializeType());

        Method declaredMethod = proxyClass.getDeclaredMethod(methodName, argTypes);
        Object result = declaredMethod.invoke(ref, getInvokeArgs(request, argTypes, serializer));

        Response.Builder builder = Response.newBuilder();
        builder.setSerializeType(request.getSerializeType());
        builder.setType(declaredMethod.getReturnType().getName());
        builder.setData(ByteString.copyFrom(serializer.encode(result, null).array()));
        Response build = builder.build();
        responseObserver.onNext(build);
        responseObserver.onCompleted();
    } catch (Exception e) {
        LOGGER.error("Invoke " + methodName + " error:", e);
        throw new SofaRpcRuntimeException(e);
    }
}
 

@Override
public void getPendingTaskForWorkflow(TaskServicePb.PendingTaskRequest req, StreamObserver<TaskServicePb.PendingTaskResponse> response) {
    try {
        Task t = taskService.getPendingTaskForWorkflow(req.getWorkflowId(), req.getTaskRefName());
        response.onNext(
                TaskServicePb.PendingTaskResponse.newBuilder()
                        .setTask(PROTO_MAPPER.toProto(t))
                        .build()
        );
        response.onCompleted();
    } catch (Exception e) {
        GRPC_HELPER.onError(response, e);
    }
}
 

@Override
public void uploadConversionAdjustments(
    UploadConversionAdjustmentsRequest request,
    StreamObserver<UploadConversionAdjustmentsResponse> responseObserver) {
  Object response = responses.remove();
  if (response instanceof UploadConversionAdjustmentsResponse) {
    requests.add(request);
    responseObserver.onNext((UploadConversionAdjustmentsResponse) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void geoServerListDatastoresCommand(
    final org.locationtech.geowave.service.grpc.protobuf.GeoServerListDatastoresCommandParametersProtos request,
    final StreamObserver<org.locationtech.geowave.service.grpc.protobuf.GeoWaveReturnTypesProtos.StringResponseProtos> responseObserver) {

  final GeoServerListDatastoresCommand cmd = new GeoServerListDatastoresCommand();
  final Map<FieldDescriptor, Object> m = request.getAllFields();
  GeoWaveGrpcServiceCommandUtil.setGrpcToCommandFields(m, cmd);

  final File configFile = GeoWaveGrpcServiceOptions.geowaveConfigFile;
  final OperationParams params = new ManualOperationParams();
  params.getContext().put(ConfigOptions.PROPERTIES_FILE_CONTEXT, configFile);

  cmd.prepare(params);

  LOGGER.info("Executing GeoServerListDatastoresCommand...");
  try {
    final String result = cmd.computeResults(params);
    final StringResponseProtos resp =
        StringResponseProtos.newBuilder().setResponseValue(result).build();
    responseObserver.onNext(resp);
    responseObserver.onCompleted();

  } catch (final Exception e) {
    LOGGER.error("Exception encountered executing command", e);
  }
}
 

@Override
public void getBlockByLimitNext(BlockLimit request,
                                StreamObserver<BlockList> responseObserver) {
    long startNum = request.getStartNum();
    long endNum = request.getEndNum();

    if (endNum > 0 && endNum > startNum && endNum - startNum <= BLOCK_LIMIT_NUM) {
        responseObserver.onNext(wallet.getBlocksByLimitNext(startNum, endNum - startNum));
    } else {
        responseObserver.onNext(null);
    }
    responseObserver.onCompleted();
}
 

@Override
public void getFrames(FrameGetFramesRequest request, StreamObserver<FrameGetFramesResponse> responseObserver) {
    responseObserver.onNext(FrameGetFramesResponse.newBuilder()
            .setFrames(
                    whiteboard.getFrames(
                            frameSearchFactory.create(
                                    jobManagerSupport.getJobManager().findJob(request.getJob()),
                                    request.getR())))
            .build());
    responseObserver.onCompleted();
}
 

@Override
public void mutateLabels(
    MutateLabelsRequest request, StreamObserver<MutateLabelsResponse> responseObserver) {
  Object response = responses.remove();
  if (response instanceof MutateLabelsResponse) {
    requests.add(request);
    responseObserver.onNext((MutateLabelsResponse) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void sayAnotherHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
  HelloReply reply =
      HelloReply.newBuilder()
          .setMessage("Hello again " + req.getName() + " " + count++)
          .build();
  responseObserver.onNext(reply);
  responseObserver.onCompleted();
}
 

@Override
public void write(WriteRequest request, StreamObserver<WriteResponse> responseObserver) {
    writeReqs.add(request);
    responseObserver.onNext(WriteResponse.getDefaultInstance());
    responseObserver.onCompleted();
    complete();
}
 

@Override
public void getCustomerExtensionSetting(
    GetCustomerExtensionSettingRequest request,
    StreamObserver<CustomerExtensionSetting> responseObserver) {
  Object response = responses.remove();
  if (response instanceof CustomerExtensionSetting) {
    requests.add(request);
    responseObserver.onNext((CustomerExtensionSetting) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void getKeywordPlanKeyword(
    GetKeywordPlanKeywordRequest request, StreamObserver<KeywordPlanKeyword> responseObserver) {
  Object response = responses.remove();
  if (response instanceof KeywordPlanKeyword) {
    requests.add(request);
    responseObserver.onNext((KeywordPlanKeyword) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void updateConnection(UpdateConnectionRequest request, StreamObserver<UpdateConnectionResponse> responseObserver) {
    if (request.getConnection().getId() == 0) {
        log.info("Cannot update a connection if ID is not present {}", request.getConnection().getId());
        responseObserver.onError(Status.INVALID_ARGUMENT.withDescription(CONNECTION_NOT_FOUND).asRuntimeException());
        return;
    }

    UpdateConnectionDTO dto = new UpdateConnectionDTO();
    dto.setId(request.getConnection().getId());
    dto.setName(request.getConnection().getName());
    dto.setConnectionUsername(request.getConnection().getConnectionUsername());
    dto.setConnectionPassword(request.getConnection().getConnectionPassword());
    dto.setLinkId(request.getConnection().getLinkId());
    dto.setConnectionType(connectionTypeService.findOne(request.getConnection().getConnectionType()));
    dto.setDetails(connectionDetailsMapper.mapToEntity(request.getConnection().getDetailsMap()));

    log.info("Updating connection {}", dto);

    ConnectionDTO createdConnection = connectionService.updateConnection(dto);

    log.debug("Updated connection {}", createdConnection);

    connectionParameterService.save(createdConnection.getLinkId(), request.getConnection().getConnectionParametersMap());
    Map<String, String> connectionParameters = connectionParameterService.getParametersByLinkId(createdConnection.getLinkId());

    log.debug("Updated connection parameters {}", connectionParameters);

    responseObserver.onNext(UpdateConnectionResponse.newBuilder()
        .setConnection(toConnectionProto(createdConnection, connectionParameters))
        .build());
    responseObserver.onCompleted();
}
 

@Override
public void getContract(BytesMessage request,
                        StreamObserver<Protocol.SmartContract> responseObserver) {
    Protocol.SmartContract contract = wallet.getContract(request);
    responseObserver.onNext(contract);
    responseObserver.onCompleted();
}
 

@Override
public void setThreadMode(HostSetThreadModeRequest request, StreamObserver<HostSetThreadModeResponse> responseObserver) {
    HostInterface host = getHostInterface(request.getHost());
    hostDao.updateThreadMode(host, request.getMode());
    responseObserver.onNext(HostSetThreadModeResponse.newBuilder().build());
    responseObserver.onCompleted();
}
 

@Override
public void mutateAdParameters(
    MutateAdParametersRequest request,
    StreamObserver<MutateAdParametersResponse> responseObserver) {
  Object response = responses.remove();
  if (response instanceof MutateAdParametersResponse) {
    requests.add(request);
    responseObserver.onNext((MutateAdParametersResponse) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void mutateCustomerFeeds(
    MutateCustomerFeedsRequest request,
    StreamObserver<MutateCustomerFeedsResponse> responseObserver) {
  Object response = responses.remove();
  if (response instanceof MutateCustomerFeedsResponse) {
    requests.add(request);
    responseObserver.onNext((MutateCustomerFeedsResponse) response);
    responseObserver.onCompleted();
  } else if (response instanceof Exception) {
    responseObserver.onError((Exception) response);
  } else {
    responseObserver.onError(new IllegalArgumentException("Unrecognized response type"));
  }
}
 

@Override
public void getService(ServiceGetServiceRequest request,
                       StreamObserver<ServiceGetServiceResponse> responseObserver) {
    try {
        responseObserver.onNext(ServiceGetServiceResponse.newBuilder()
                .setService(whiteboard.getService(request.getName()))
                .build());
        responseObserver.onCompleted();
    } catch (EmptyResultDataAccessException e) {
        responseObserver.onError(Status.NOT_FOUND
                .withDescription(e.getMessage())
                .withCause(e)
                .asRuntimeException());
    }
}
 

@Test
public void grpclbThenNameResolutionFails() {
  InOrder inOrder = inOrder(helper, subchannelPool);
  // Go to GRPCLB first
  List<EquivalentAddressGroup> grpclbBalancerList = createResolvedBalancerAddresses(1);
  deliverResolvedAddresses(Collections.<EquivalentAddressGroup>emptyList(), grpclbBalancerList);

  verify(helper).createOobChannel(eq(grpclbBalancerList.get(0)), eq(lbAuthority(0)));
  assertEquals(1, fakeOobChannels.size());
  ManagedChannel oobChannel = fakeOobChannels.poll();
  verify(mockLbService).balanceLoad(lbResponseObserverCaptor.capture());
  StreamObserver<LoadBalanceResponse> lbResponseObserver = lbResponseObserverCaptor.getValue();

  // Let name resolution fail before round-robin list is ready
  Status error = Status.NOT_FOUND.withDescription("www.google.com not found");
  deliverNameResolutionError(error);

  inOrder.verify(helper).updateBalancingState(eq(TRANSIENT_FAILURE), pickerCaptor.capture());
  RoundRobinPicker picker = (RoundRobinPicker) pickerCaptor.getValue();
  assertThat(picker.dropList).isEmpty();
  assertThat(picker.pickList).containsExactly(new ErrorEntry(error));
  assertFalse(oobChannel.isShutdown());

  // Simulate receiving LB response
  List<ServerEntry> backends = Arrays.asList(
      new ServerEntry("127.0.0.1", 2000, "TOKEN1"),
      new ServerEntry("127.0.0.1", 2010, "TOKEN2"));
  lbResponseObserver.onNext(buildInitialResponse());
  lbResponseObserver.onNext(buildLbResponse(backends));

  inOrder.verify(subchannelPool).takeOrCreateSubchannel(
      eq(new EquivalentAddressGroup(backends.get(0).addr, LB_BACKEND_ATTRS)),
      any(Attributes.class));
  inOrder.verify(subchannelPool).takeOrCreateSubchannel(
      eq(new EquivalentAddressGroup(backends.get(1).addr, LB_BACKEND_ATTRS)),
      any(Attributes.class));
}
 

/**
 * (EDS response caching behavior) An endpoint watcher is registered for a cluster that already
 * has some other endpoint watchers watching on. Endpoint information received previously is
 * in local cache and notified to the new watcher immediately.
 */
@Test
public void watchEndpointsForClusterAlreadyBeingWatched() {
  EndpointWatcher watcher1 = mock(EndpointWatcher.class);
  xdsClient.watchEndpointData("cluster-foo.googleapis.com", watcher1);
  StreamObserver<DiscoveryResponse> responseObserver = responseObservers.poll();
  StreamObserver<DiscoveryRequest> requestObserver = requestObservers.poll();

  // Client sends first EDS request.
  verify(requestObserver)
      .onNext(eq(buildDiscoveryRequest(NODE, "", "cluster-foo.googleapis.com",
          XdsClientImpl.ADS_TYPE_URL_EDS, "")));

  assertThat(fakeClock.getPendingTasks(EDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER)).hasSize(1);

  // Management server sends back an EDS response containing ClusterLoadAssignments for
  // some cluster not requested.
  List<Any> clusterLoadAssignments = ImmutableList.of(
      Any.pack(buildClusterLoadAssignment("cluster-foo.googleapis.com",
          ImmutableList.of(
              buildLocalityLbEndpoints("region1", "zone1", "subzone1",
                  ImmutableList.of(
                      buildLbEndpoint("192.168.0.1", 8080, HealthStatus.HEALTHY, 2)),
                  1, 0)),
          ImmutableList.<Policy.DropOverload>of())));

  DiscoveryResponse response =
      buildDiscoveryResponse("0", clusterLoadAssignments,
          XdsClientImpl.ADS_TYPE_URL_EDS, "0000");
  responseObserver.onNext(response);

  assertThat(fakeClock.getPendingTasks(EDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER)).isEmpty();

  // Client sent an ACK EDS request.
  verify(requestObserver)
      .onNext(eq(buildDiscoveryRequest(NODE, "0", "cluster-foo.googleapis.com",
          XdsClientImpl.ADS_TYPE_URL_EDS, "0000")));

  ArgumentCaptor<EndpointUpdate> endpointUpdateCaptor1 = ArgumentCaptor.forClass(null);
  verify(watcher1).onEndpointChanged(endpointUpdateCaptor1.capture());
  EndpointUpdate endpointUpdate1 = endpointUpdateCaptor1.getValue();
  assertThat(endpointUpdate1.getClusterName()).isEqualTo("cluster-foo.googleapis.com");
  assertThat(endpointUpdate1.getDropPolicies()).isEmpty();
  assertThat(endpointUpdate1.getLocalityLbEndpointsMap())
      .containsExactly(
          new Locality("region1", "zone1", "subzone1"),
          new LocalityLbEndpoints(
              ImmutableList.of(
                  new LbEndpoint("192.168.0.1", 8080,
                      2, true)), 1, 0));

  // A second endpoint watcher is registered for endpoints in the same cluster.
  EndpointWatcher watcher2 = mock(EndpointWatcher.class);
  xdsClient.watchEndpointData("cluster-foo.googleapis.com", watcher2);

  // Cached endpoint information is notified to the new watcher immediately, without sending
  // another EDS request.
  ArgumentCaptor<EndpointUpdate> endpointUpdateCaptor2 = ArgumentCaptor.forClass(null);
  verify(watcher2).onEndpointChanged(endpointUpdateCaptor2.capture());
  EndpointUpdate endpointUpdate2 = endpointUpdateCaptor2.getValue();
  assertThat(endpointUpdate2.getClusterName()).isEqualTo("cluster-foo.googleapis.com");
  assertThat(endpointUpdate2.getDropPolicies()).isEmpty();
  assertThat(endpointUpdate2.getLocalityLbEndpointsMap())
      .containsExactly(
          new Locality("region1", "zone1", "subzone1"),
          new LocalityLbEndpoints(
              ImmutableList.of(
                  new LbEndpoint("192.168.0.1", 8080,
                      2, true)), 1, 0));

  verifyNoMoreInteractions(requestObserver);
  assertThat(fakeClock.getPendingTasks(EDS_RESOURCE_FETCH_TIMEOUT_TASK_FILTER)).isEmpty();
}
 

/**
 * Tests client per-message compression for streaming calls. The Java API does not support
 * inspecting a message's compression level, so this is primarily intended to run against a gRPC
 * C++ server.
 */
public void clientCompressedStreaming(boolean probe) throws Exception {
  final StreamingInputCallRequest expectCompressedRequest =
      StreamingInputCallRequest.newBuilder()
          .setExpectCompressed(BoolValue.newBuilder().setValue(true))
          .setPayload(Payload.newBuilder().setBody(ByteString.copyFrom(new byte[27182])))
          .build();
  final StreamingInputCallRequest expectUncompressedRequest =
      StreamingInputCallRequest.newBuilder()
          .setExpectCompressed(BoolValue.newBuilder().setValue(false))
          .setPayload(Payload.newBuilder().setBody(ByteString.copyFrom(new byte[45904])))
          .build();
  final StreamingInputCallResponse goldenResponse =
      StreamingInputCallResponse.newBuilder().setAggregatedPayloadSize(73086).build();

  StreamRecorder<StreamingInputCallResponse> responseObserver = StreamRecorder.create();
  StreamObserver<StreamingInputCallRequest> requestObserver =
      asyncStub.streamingInputCall(responseObserver);

  if (probe) {
    // Send a non-compressed message with expectCompress=true. Servers supporting this test case
    // should return INVALID_ARGUMENT.
    requestObserver.onNext(expectCompressedRequest);
    responseObserver.awaitCompletion(operationTimeoutMillis(), TimeUnit.MILLISECONDS);
    Throwable e = responseObserver.getError();
    assertNotNull("expected INVALID_ARGUMENT", e);
    assertEquals(Status.INVALID_ARGUMENT.getCode(), Status.fromThrowable(e).getCode());
  }

  // Start a new stream
  responseObserver = StreamRecorder.create();
  @SuppressWarnings("unchecked")
  ClientCallStreamObserver<StreamingInputCallRequest> clientCallStreamObserver =
      (ClientCallStreamObserver)
          asyncStub.withCompression("gzip").streamingInputCall(responseObserver);
  clientCallStreamObserver.setMessageCompression(true);
  clientCallStreamObserver.onNext(expectCompressedRequest);
  clientCallStreamObserver.setMessageCompression(false);
  clientCallStreamObserver.onNext(expectUncompressedRequest);
  clientCallStreamObserver.onCompleted();
  responseObserver.awaitCompletion();
  assertSuccess(responseObserver);
  assertEquals(goldenResponse, responseObserver.firstValue().get());
}