Method transferFlowFiles

libminifi/src/sitetosite/SiteToSiteClient.cpp:115–204

}

bool SiteToSiteClient::transferFlowFiles(const std::shared_ptr<core::ProcessContext> &context, const std::shared_ptr<core::ProcessSession> &session) {
  auto flow = session->get();

  std::shared_ptr<Transaction> transaction = nullptr;

  if (!flow) {
    return false;
  }

  if (peer_state_ != READY) {
    if (!bootstrap())
      return false;
  }

  if (peer_state_ != READY) {
    context->yield();
    tearDown();
    throw Exception(SITE2SITE_EXCEPTION, "Can not establish handshake with peer");
  }

  // Create the transaction
  transaction = createTransaction(SEND);
  if (transaction == nullptr) {
    context->yield();
    tearDown();
    throw Exception(SITE2SITE_EXCEPTION, "Can not create transaction");
  }
  utils::Identifier transactionID = transaction->getUUID();

  bool continueTransaction = true;
  uint64_t startSendingNanos = utils::timeutils::getTimeNano();

  try {
    while (continueTransaction) {
      uint64_t startTime = utils::timeutils::getTimeMillis();
      std::string payload;
      DataPacket packet(getLogger(), transaction, flow->getAttributes(), payload);

      int16_t resp = send(transactionID, &packet, flow, session);
      if (resp == -1) {
        throw Exception(SITE2SITE_EXCEPTION, "Send Failed");
      }

      logger_->log_debug("Site2Site transaction %s send flow record %s", transactionID.to_string(), flow->getUUIDStr());
      if (resp == 0) {
        uint64_t endTime = utils::timeutils::getTimeMillis();
        std::string transitUri = peer_->getURL() + "/" + flow->getUUIDStr();
        std::string details = "urn:nifi:" + flow->getUUIDStr() + "Remote Host=" + peer_->getHostName();
        session->getProvenanceReporter()->send(flow, transitUri, details, endTime - startTime, false);
      }
      session->remove(flow);

      uint64_t transferNanos = utils::timeutils::getTimeNano() - startSendingNanos;
      if (transferNanos > _batchSendNanos)
        break;

      flow = session->get();