Method poll

libminifi/src/Connection.cpp:181–224

}

std::shared_ptr<core::FlowFile> Connection::poll(std::set<std::shared_ptr<core::FlowFile>> &expiredFlowRecords) {
  std::lock_guard<std::mutex> lock(mutex_);

  while (!queue_.empty()) {
    std::shared_ptr<core::FlowFile> item = queue_.front();
    queue_.pop();
    queued_data_size_ -= item->getSize();

    if (expired_duration_ > 0) {
      // We need to check for flow expiration
      if (utils::timeutils::getTimeMillis() > (item->getEntryDate() + expired_duration_)) {
        // Flow record expired
        expiredFlowRecords.insert(item);
        logger_->log_debug("Delete flow file UUID %s from connection %s, because it expired", item->getUUIDStr(), name_);
      } else {
        // Flow record not expired
        if (item->isPenalized()) {
          // Flow record was penalized
          queue_.push(item);
          queued_data_size_ += item->getSize();
          break;
        }
        std::shared_ptr<Connectable> connectable = std::static_pointer_cast<Connectable>(shared_from_this());
        item->setConnection(connectable);
        logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr(), name_);
        return item;
      }
    } else {
      // Flow record not expired
      if (item->isPenalized()) {
        // Flow record was penalized
        queue_.push(item);
        queued_data_size_ += item->getSize();
        break;
      }
      std::shared_ptr<Connectable> connectable = std::static_pointer_cast<Connectable>(shared_from_this());
      item->setConnection(connectable);
      logger_->log_debug("Dequeue flow file UUID %s from connection %s", item->getUUIDStr(), name_);
      return item;
    }
  }

  return NULL;
}

void Connection::drain(bool delete_permanently) {
  std::lock_guard<std::mutex> lock(mutex_);