Method heartbeat

libminifi/src/c2/ControllerSocketProtocol.cpp:261–302

}

int16_t ControllerSocketProtocol::heartbeat(const C2Payload &payload) {
  if (server_socket_ == nullptr)
    return 0;
  const std::vector<C2ContentResponse> &content = payload.getContent();
  for (const auto &pc : payload.getNestedPayloads()) {
    if (pc.getLabel() == "flowInfo" || pc.getLabel() == "metrics") {
      for (const auto &metrics_payload : pc.getNestedPayloads()) {
        if (metrics_payload.getLabel() == "QueueMetrics" || metrics_payload.getLabel() == "queues") {
          for (const auto &queue_metrics : metrics_payload.getNestedPayloads()) {
            auto metric_content = queue_metrics.getContent();
            for (const auto &payload_content : queue_metrics.getContent()) {
              uint64_t size = 0;
              uint64_t max = 0;
              for (auto content : payload_content.operation_arguments) {
                if (content.first == "datasize") {
                  size = std::stol(content.second.to_string());
                } else if (content.first == "datasizemax") {
                  max = std::stol(content.second.to_string());
                }
              }
              std::lock_guard<std::mutex> lock(controller_mutex_);
              if (size >= max) {
                queue_full_[payload_content.name] = true;
              } else {
                queue_full_[payload_content.name] = false;
              }
              queue_size_[payload_content.name] = size;
              queue_max_[payload_content.name] = max;
            }
          }
        }
      }
    }
  }

  parse_content(content);

  std::vector<uint8_t> buffer;
  buffer.resize(1024);

  return 0;
}

}  // namespace c2
}  // namespace minifi