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

@Override
public void findAllInputsOutputs(
    FindAllInputsOutputs request,
    StreamObserver<FindAllInputsOutputs.Response> responseObserver) {
  QPSCountResource.inc();
  try {
    if (request.getItemsCount() == 0) {
      throw new ModelDBException("Items not specified", Code.INVALID_ARGUMENT);
    }
    try (RequestLatencyResource latencyResource =
        new RequestLatencyResource(ModelDBAuthInterceptor.METHOD_NAME.get())) {
      FindAllInputsOutputs.Response response = lineageDAO.findAllInputsOutputs(request);
      responseObserver.onNext(response);
      responseObserver.onCompleted();
    }
  } catch (Exception e) {
    ModelDBUtils.observeError(
        responseObserver, e, FindAllInputsOutputs.Response.getDefaultInstance());
  }
}
 

/**
 * Implement a BI-DIRECTIONAL STREAMING operation
 */
@Override
public StreamObserver<HelloRequest> streamGreet(StreamObserver<HelloResponse> responseObserver) {
    // Notice how the programming model completely changes
    return new StreamObserver<HelloRequest>() {
        @Override
        public void onNext(HelloRequest request) {
            String name = request.getName();
            responseObserver.onNext(HelloResponse.newBuilder().setMessage("Welcome " + name).build());
        }

        @Override
        public void onError(Throwable t) {
            responseObserver.onCompleted();
        }

        @Override
        public void onCompleted() {
            responseObserver.onCompleted();
        }
    };
}
 

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

@Override
public void withdrawBalance(Contract.WithdrawBalanceContract request,
                            StreamObserver<Transaction> responseObserver) {
    try {
        responseObserver.onNext(
                createTransactionWrapper(request, ContractType.WithdrawBalanceContract)
                        .getInstance());
    } catch (ContractValidateException e) {
        responseObserver
                .onNext(null);
        logger.debug("ContractValidateException: {}", e.getMessage());
    }
    responseObserver.onCompleted();
}
 

@Override
public void getTasks(DeptGetTasksRequest request, StreamObserver<DeptGetTasksResponse> responseObserver) {
    PointDetail deptConfig = departmentManager.getDepartmentConfigDetail(request.getDepartment().getId());
    TaskSeq tasks = whiteboard.getTasks(deptConfig, deptConfig);
    TaskSeq taskSeq = TaskSeq.newBuilder().addAllTasks(tasks.getTasksList()).build();
    responseObserver.onNext(DeptGetTasksResponse.newBuilder().setTasks(taskSeq).build());
    responseObserver.onCompleted();
}
 

@Override
public void open(SessionProto.Session.Open.Req request, StreamObserver<SessionProto.Session.Open.Res> responseObserver) {
    try {
        String keyspace = request.getKeyspace();
        Session session = requestOpener.open(request);
        String sessionId = keyspace + UUID.randomUUID().toString();
        openSessions.put(sessionId, session);
        transactionListeners.put(sessionId, new HashSet<>());
        responseObserver.onNext(SessionProto.Session.Open.Res.newBuilder().setSessionId(sessionId).build());
        responseObserver.onCompleted();
    } catch (RuntimeException e) {
        LOG.error("An error has occurred", e);
        responseObserver.onError(ResponseBuilder.exception(e));
    }
}
 

@Override
public void subscription(SubscriptionReq request, StreamObserver<SubscriptionsResp> responseObserver) {
    SubscriptionsResp resp = SubscriptionsResp.newBuilder()
                                              .addMetricNames("first")
                                              .addMetricNames("second")
                                              .build();
    responseObserver.onNext(resp);
    responseObserver.onCompleted();
}
 

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

@Override
public void onTxEvent(GrpcTxEvent request, StreamObserver<GrpcAck> responseObserver) {
  events.offer(new TxEvent(
      EventType.valueOf(request.getType()),
      request.getGlobalTxId(),
      request.getLocalTxId(),
      request.getParentTxId(),
      request.getCompensationMethod(),
      request.getForwardTimeout(),
      request.getRetryMethod(),
      request.getForwardRetries(),
      request.getForwardTimeout(),
      request.getReverseRetries(),
      request.getReverseTimeout(),
      request.getRetryDelayInMilliseconds(),
      new String(request.getPayloads().toByteArray())));

  sleep();

  if (EventType.TxAbortedEvent.name().equals(request.getType())) {
    this.responseObserver.onNext(GrpcCompensateCommand
        .newBuilder()
        .setGlobalTxId(request.getGlobalTxId())
        .build());
  }

  if ("TxStartedEvent".equals(request.getType()) && request.getCompensationMethod().equals("reject")) {
    responseObserver.onNext(GrpcAck.newBuilder().setAborted(true).build());
  } else {
    responseObserver.onNext(GrpcAck.newBuilder().setAborted(false).build());
  }

  responseObserver.onCompleted();
}
 

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

@Override
public void staggerFrames(JobStaggerFramesRequest request,
                          StreamObserver<JobStaggerFramesResponse> responseObserver) {
    setupJobData(request.getJob());
    manageQueue.execute(
            new DispatchStaggerFrames(job, request.getRange(), request.getStagger(), jobManagerSupport));
    responseObserver.onNext(JobStaggerFramesResponse.newBuilder().build());
    responseObserver.onCompleted();
}
 

@Override
public void sayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
    HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).build();
    // 使用响应监视器的onNext方法返回HelloReply
    responseObserver.onNext(reply);
    // 使用onCompleted方法指定本次调用已经完成
    responseObserver.onCompleted();
}
 

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

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

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

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

@Override
public void getArtifacts(
    GetArtifacts request, StreamObserver<GetArtifacts.Response> responseObserver) {
  QPSCountResource.inc();
  try (RequestLatencyResource latencyResource =
      new RequestLatencyResource(ModelDBAuthInterceptor.METHOD_NAME.get())) {

    if (request.getId().isEmpty()) {
      String errorMessage = "Project ID not found in GetArtifacts request";
      ModelDBUtils.logAndThrowError(
          errorMessage,
          Code.INVALID_ARGUMENT_VALUE,
          Any.pack(GetArtifacts.Response.getDefaultInstance()));
    }

    // Validate if current user has access to the entity or not
    roleService.validateEntityUserWithUserInfo(
        ModelDBServiceResourceTypes.PROJECT, request.getId(), ModelDBServiceActions.READ);

    List<Artifact> artifactList = projectDAO.getProjectArtifacts(request.getId());
    responseObserver.onNext(
        GetArtifacts.Response.newBuilder().addAllArtifacts(artifactList).build());
    responseObserver.onCompleted();

  } catch (Exception e) {
    ModelDBUtils.observeError(responseObserver, e, GetArtifacts.Response.getDefaultInstance());
  }
}
 

@Override
public void asynSayHello(HelloRequest req, StreamObserver<HelloReply> responseObserver) {
  System.out.println("server gets msg: id = " + req.getId());
  try {
    Random random = new Random();
    int duration = random.nextInt(2) + 1;
    TimeUnit.SECONDS.sleep(duration);
  } catch (InterruptedException e) {
    logger.error(e.getMessage(), e);
  }

  HelloReply reply = HelloReply.newBuilder().setMessage("Hello " + req.getName()).setNo(req.getId()).build();
  responseObserver.onNext(reply);
  responseObserver.onCompleted();
}
 

@Override
public void delete(ActionDeleteRequest request, StreamObserver<ActionDeleteResponse> responseObserver) {
    Action requestAction = request.getAction();
    ActionEntity existingAction = filterManager.getAction(requestAction.getId());
    FilterEntity filterEntity = filterManager.getFilter(existingAction);
    ActionEntity actionToDelete = ActionEntity.build(filterEntity, requestAction, requestAction.getId());
    filterManager.deleteAction(actionToDelete);
    responseObserver.onNext(ActionDeleteResponse.newBuilder().build());
    responseObserver.onCompleted();
}
 

public void setOrder(FilterSetOrderRequest request, StreamObserver<FilterSetOrderResponse> responseObserver) {
    FilterEntity filter = getFilterEntity(request.getFilter());
    filterManager.setFilterOrder(filter, (double) request.getOrder());
    responseObserver.onNext(FilterSetOrderResponse.newBuilder().build());
    responseObserver.onCompleted();
}