com.google.protobuf.RpcCallback#run ( )源码实例Demo

@Override
public void getAllWorkerResource(RpcController controller, PrimitiveProtos.NullProto request,
                                 RpcCallback<TajoMasterProtocol.WorkerResourcesRequest> done) {

  TajoMasterProtocol.WorkerResourcesRequest.Builder builder =
      TajoMasterProtocol.WorkerResourcesRequest.newBuilder();
  Collection<Worker> workers = context.getResourceManager().getWorkers().values();

  for(Worker worker: workers) {
    WorkerResource resource = worker.getResource();

    TajoMasterProtocol.WorkerResourceProto.Builder workerResource =
        TajoMasterProtocol.WorkerResourceProto.newBuilder();

    workerResource.setHost(worker.getHostName());
    workerResource.setPeerRpcPort(worker.getPeerRpcPort());
    workerResource.setInfoPort(worker.getHttpPort());
    workerResource.setQueryMasterPort(worker.getQueryMasterPort());
    workerResource.setMemoryMB(resource.getMemoryMB());
    workerResource.setDiskSlots(resource.getDiskSlots());
    workerResource.setQueryMasterPort(worker.getQueryMasterPort());

    builder.addWorkerResources(workerResource);
  }
  done.run(builder.build());
}
 

@Override
public void heartbeat(
    RpcController controller,
    TajoHeartbeatRequest request, RpcCallback<TajoHeartbeatResponse> done) {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Received QueryHeartbeat:" + new WorkerConnectionInfo(request.getConnectionInfo()));
  }

  TajoHeartbeatResponse.ResponseCommand command;

  QueryManager queryManager = context.getQueryJobManager();
  command = queryManager.queryHeartbeat(request);

  TajoHeartbeatResponse.Builder builder = TajoHeartbeatResponse.newBuilder();
  builder.setHeartbeatResult(TRUE);
  if(command != null) {
    builder.setResponseCommand(command);
  }

  done.run(builder.build());
}
 

@Override
public void getTask(RpcController controller, TajoWorkerProtocol.GetTaskRequestProto request,
                    RpcCallback<TajoWorkerProtocol.QueryUnitRequestProto> done) {
  try {
    ExecutionBlockId ebId = new ExecutionBlockId(request.getExecutionBlockId());
    QueryMasterTask queryMasterTask = workerContext.getQueryMaster().getQueryMasterTask(ebId.getQueryId());

    if(queryMasterTask == null || queryMasterTask.isStopped()) {
      done.run(LazyTaskScheduler.stopTaskRunnerReq);
    } else {
      ContainerId cid =
          queryMasterTask.getQueryTaskContext().getResourceAllocator().makeContainerId(request.getContainerId());
      LOG.debug("getTask:" + cid + ", ebId:" + ebId);
      queryMasterTask.handleTaskRequestEvent(new TaskRequestEvent(cid, ebId, done));
    }
  } catch (Exception e) {
    LOG.error(e.getMessage(), e);
  }
}
 

/**
 * Registers a cancellation callback. This will always be called as part of
 * this RPC, and always at most once; either when the client asks to cancel
 * it, or when the RPC finishes (regardless of error case).
 *
 * This callback will be called outside normal locks on ServerRpcController
 * state, i.e., not within a block synchronised on statusLock.
 */
@Override
public void notifyOnCancel(final RpcCallback<Object> callback) {
  RpcCallback<Object> runCallback = null;
  synchronized (statusLock) {
    if (cancelCallback != null) {
      throw new IllegalStateException("Must only be called once per request.");
    } else {
      cancelCallback = callback;
      if (cancelled || complete) {
        runCallback = cancelCallback;
      }
    }
  }
  if (runCallback != null) {
    runCallback.run(null);
  }
}
 

@Override
public void heartbeat(
    RpcController controller,
    TajoMasterProtocol.TajoHeartbeat request, RpcCallback<TajoMasterProtocol.TajoHeartbeatResponse> done) {
  if(LOG.isDebugEnabled()) {
    LOG.debug("Received QueryHeartbeat:" + request.getTajoWorkerHost() + ":" + request.getTajoQueryMasterPort());
  }

  TajoMasterProtocol.TajoHeartbeatResponse.ResponseCommand command = null;

  QueryJobManager queryJobManager = context.getQueryJobManager();
  command = queryJobManager.queryHeartbeat(request);

  TajoMasterProtocol.TajoHeartbeatResponse.Builder builder = TajoMasterProtocol.TajoHeartbeatResponse.newBuilder();
  builder.setHeartbeatResult(BOOL_TRUE);
  if(command != null) {
    builder.setResponseCommand(command);
  }

  builder.setClusterResourceSummary(context.getResourceManager().getClusterResourceSummary());
  done.run(builder.build());
}
 

@Override
public void version(
    final RpcController controller,
    final VersionRequest request,
    final RpcCallback<VersionResponse> done) {
  done.run(VersionResponse.newBuilder().addAllVersionInfo(VersionUtils.getVersionInfo()).build());
}
 

@Override
public void delay(RpcController controller, EchoMessage request,
                  RpcCallback<EchoMessage> done) {
  try {
    Thread.sleep(3000);
  } catch (InterruptedException e) {
    LOG.error(e.getMessage());
  }

  done.run(request);
}
 

@Override
public void sum(RpcController controller, SumRequest request,
                RpcCallback<SumResponse> done) {

  SumResponse response = SumResponse.newBuilder().setResult(
      request.getX1()+request.getX2()+request.getX3()+request.getX4()
  ).build();
  done.run(response);
}
 

@Override
public void bidding(RpcController controller, BiddingReq request, RpcCallback<BiddingRsp> done) {
	try {
		done.run(bidding(controller, request));
	} catch (ServiceException ex) {
	}
}
 

@Override
public void statusUpdate(RpcController controller, TaskStatusProto request,
                         RpcCallback<PrimitiveProtos.NullProto> done) {
  QueryId queryId = new QueryId(request.getId().getTaskId().getExecutionBlockId().getQueryId());
  TaskAttemptId attemptId = new TaskAttemptId(request.getId());
  QueryMasterTask queryMasterTask = queryMaster.getQueryMasterTask(queryId);

  if (queryMasterTask != null) {
    Stage sq = queryMasterTask.getQuery().getStage(attemptId.getTaskId().getExecutionBlockId());
    Task task = sq.getTask(attemptId.getTaskId());
    TaskAttempt attempt = task.getAttempt(attemptId.getId());

    if(LOG.isDebugEnabled()){
      LOG.debug(String.format("Task State: %s, Attempt State: %s", task.getState().name(), attempt.getState().name()));
    }

    if (request.getState() == TajoProtos.TaskAttemptState.TA_KILLED) {
      LOG.warn(attemptId + " Killed");
      attempt.handle(
          new TaskAttemptEvent(new TaskAttemptId(request.getId()), TaskAttemptEventType.TA_LOCAL_KILLED));
    } else {
      queryMasterTask.getEventHandler().handle(
          new TaskAttemptStatusUpdateEvent(new TaskAttemptId(request.getId()), request));
    }

  }
  done.run(TajoWorker.NULL_PROTO);
}
 

@Override
public void doneExecutionBlock(
    RpcController controller, ExecutionBlockReport request,
    RpcCallback<PrimitiveProtos.NullProto> done) {
  QueryMasterTask queryMasterTask = queryMaster.getQueryMasterTask(new QueryId(request.getEbId().getQueryId()));
  if (queryMasterTask != null) {
    ExecutionBlockId ebId = new ExecutionBlockId(request.getEbId());
    queryMasterTask.getEventHandler().handle(new StageShuffleReportEvent(ebId, request));
  }
  done.run(TajoWorker.NULL_PROTO);
}
 

@Override
public void executeQuery(RpcController controller,
                         QueryExecutionRequest request,
                         RpcCallback<PrimitiveProtos.NullProto> done) {

  QueryId queryId = new QueryId(request.getQueryId());
  LOG.info("Receive executeQuery request:" + queryId);
  queryMaster.handle(new QueryStartEvent(queryId,
      new Session(request.getSession()),
      new QueryContext(workerContext.getQueryMaster().getContext().getConf(),
          request.getQueryContext()), request.getExprInJson().getValue(),
      request.getLogicalPlanJson().getValue(), new NodeResource(request.getAllocation().getResource())));
  done.run(TajoWorker.NULL_PROTO);
}
 

@Override
public void releaseWorkerResource(RpcController controller,
                                       TajoMasterProtocol.WorkerResourceReleaseRequest request,
                                       RpcCallback<PrimitiveProtos.BoolProto> done) {
  List<YarnProtos.ContainerIdProto> containerIds = request.getContainerIdsList();

  for(YarnProtos.ContainerIdProto eachContainer: containerIds) {
    context.getResourceManager().releaseWorkerResource(eachContainer);
  }
  done.run(BOOL_TRUE);
}
 

@Override
public void removeServerCache(RpcController controller, RemoveServerCacheRequest request,
    RpcCallback<RemoveServerCacheResponse> done) {
  ImmutableBytesPtr tenantId = null;
  if (request.hasTenantId()) {
    tenantId = new ImmutableBytesPtr(request.getTenantId().toByteArray());
  }
  TenantCache tenantCache = GlobalCache.getTenantCache(this.env, tenantId);
  tenantCache.removeServerCache(new ImmutableBytesPtr(request.getCacheId().toByteArray()));
  RemoveServerCacheResponse.Builder responseBuilder = RemoveServerCacheResponse.newBuilder();
  responseBuilder.setReturn(true);
  RemoveServerCacheResponse result = responseBuilder.build();
  done.run(result);
}
 

@Override
public void fatalError(RpcController controller, TajoWorkerProtocol.TaskFatalErrorReport report,
                       RpcCallback<PrimitiveProtos.BoolProto> done) {
  try {
    QueryMasterTask queryMasterTask = queryMaster.getQueryMasterTask(
        new QueryId(report.getId().getQueryUnitId().getExecutionBlockId().getQueryId()));
    queryMasterTask.getEventHandler().handle(new TaskFatalErrorEvent(report));
    done.run(TajoWorker.TRUE_PROTO);
  } catch (Exception e) {
    LOG.error(e.getMessage(), e);
    done.run(TajoWorker.FALSE_PROTO);
  }
}
 

private static void callbackWithNull(RpcCallback<Message> done) {
  if (done != null) {
    done.run(null);
  }
}
 

@Override
public void queryByRowKey(RpcController controller, DataProtos.DataQueryRequest request, RpcCallback<DataQueryResponse> done) {
    DataProtos.DataQueryResponse response = null;
    try {
        String rowKey = request.getRowKey();
        String regionStartKey = Bytes.toString(this.env.getRegion().getRegionInfo().getStartKey());

        if (request.getIsSalting()) {                       // 如果加盐过则在key前添加盐值
            String startSalt = null;
            if (null != regionStartKey && !regionStartKey.isEmpty()) {
                startSalt = regionStartKey.split("_")[0];   // 加盐的方式为盐值+"_"，所以取_前面的
            }
            if (null != startSalt) {
                if (null != rowKey) {
                    rowKey = startSalt + "_" + rowKey;
                }
            }
        }
        if (StrUtils.isEmpty(rowKey)) return;
        Get get = new Get(Bytes.toBytes(rowKey));
        Result result = this.env.getRegion().get(get);

        DataProtos.DataQueryResponse.Builder responseBuilder = DataProtos.DataQueryResponse.newBuilder();
        DataProtos.DataQueryResponse.Row.Builder rowBuilder = DataProtos.DataQueryResponse.Row.newBuilder();

        if (result != null && !result.isEmpty()) {
            List<KeyValue> list = result.list();
            if (null != list && !list.isEmpty()) {
                rowBuilder.setRowKey(ByteString.copyFrom(list.get(0).getRow()));
                for (KeyValue kv : list) {
                    queryBuilder(rowBuilder, ByteString.copyFrom(kv.getFamily()), ByteString.copyFrom(kv.getQualifier()), ByteString.copyFrom(kv.getRow()), ByteString.copyFrom(kv.getValue()));
                }
            }
        }
        responseBuilder.addRowList(rowBuilder);
        response = responseBuilder.build();
    } catch (IOException ignored) {
        ResponseConverter.setControllerException(controller, ignored);
    }
    done.run(response);
}
 

@Override
public void ping(RpcController controller,
                 TajoIdProtos.QueryUnitAttemptIdProto attemptId,
                 RpcCallback<PrimitiveProtos.BoolProto> done) {
  done.run(TajoWorker.TRUE_PROTO);
}
 

@Override
public void echo(RpcController controller, EchoMessage request,
                 RpcCallback<EchoMessage> done) {

  done.run(request);
}
 

@Override
public void echo(RpcController controller, EchoMessage request,
                 RpcCallback<EchoMessage> done) {

  done.run(request);
}