Блог

У меня есть эти два вызова gRPC:

 service Console {
  rpc JoinNetwork(keyvaluestore.Request) returns (ClusterInfo) {}
  rpc WatchHealth(HealthCheckRequest) returns (stream HealthCheckResponse) {}
}
 

Которые используются двумя объектами: мастернодой и узлом хранения. Когда узел хранения запускается в первый раз, он вызывает JoinNetwork() мастер-узел, который возвращает некоторую информацию узлу хранения.

Затем я хочу вызвать WatchHealth() мастерноду, которая выполняет пинг узла хранения, который в ответ постоянно отправляет сообщение «сердцебиение» на сервер до тех пор, пока он не выйдет из строя.

Единственный способ, которым я могу придумать объединение этих двух вызовов вместе, — это вызов WatchHealth внутри серверной реализации JoinNetwork. Но я действительно хочу вызывать WatchHealth только после возврата JoinNetwork, что я не уверен, как это сделать.

JoinNetwork (клиент):

 bool ConsoleClient::JoinNetwork(const stringamp; my_addr, ClusterInfo* cinfo) {
  ClientContext ctx;
  Request req;
  Status status = stub_->JoinNetwork(amp;ctx, req, cinfo);
  if (status.ok()) {
    log("[C: JoinNetwork] Successfully joined cluster "   to_string(cinfo->cid()), VB);
    return true;
  } else {
    log("[C: JoinNetwork] Error "   to_string(status.error_code())   ": "   status.error_message(), VB);
    return false;
  }
}
 

Присоединиться к сети (серверу):

 Status ConsoleServiceImpl::JoinNetwork(ServerContext* ctx, const Request* req, ClusterInfo* res) {
  MLOCK.lock();
  int cid = (NUM_NODES  ) % K;
  CDIR[cid].insert(req->addr());
  if (CPRIMES.find(cid) == CPRIMES.end()) {
    CPRIMES.insert({ cid, req->addr() }); // If no primary exists, make it the primary
  }
  res->set_cid(cid);
  res->set_sid(assign_sid(cid));
  res->set_caddrs(pack_addrs(CDIR[cid]));
  res->set_primary(CPRIMES[cid]);
  log("[S: JoinNetwork] Storage node "   req->addr()   " has joined cluster "   to_string(cid), VB);
  MLOCK.unlock();

  return Status::OK;
}
 

WatchHealth (клиент):

 bool ConsoleClient::WatchHealth(const stringamp; addr) {
  ClientContext ctx;
  auto deadline = chrono::system_clock::now()   chrono::milliseconds(500);
  ctx.set_deadline(deadline);
  HealthCheckRequest req;
  HealthCheckResponse res;
  unique_ptr<ClientReader<HealthCheckResponse>> reader(stub_->WatchHealth(amp;ctx, req));
  while (reader->Read(amp;res));
  Status status = reader->Finish();
  if (status.ok()) {
    log("[C: WatchHealth] Storage node "   addr   " shutdown gracefully", VB);
    return true;
  } else {
    log("[C: WatchHealth] Storage node "   addr   " crashed", VB);
    return false;
  }
}
 

WatchHealth (сервер):

 Status ConsoleServiceImpl::WatchHealth(ServerContext* ctx, const HealthCheckRequest* req, ServerWriter<HealthCheckResponse>* writer) {
  HealthCheckResponse res;
  while (true) {
    res.set_status(console::HealthCheckResponse::SERVING);
    writer->Write(res);
  }
  res.set_status(console::HealthCheckResponse::SHUTTING_DOWN);
  writer->Write(res);
  return Status::OK;
}