Method processOnTrigger

extensions/sql/processors/ExecuteSQL.cpp:90–116

}

void ExecuteSQL::processOnTrigger(core::ProcessSession& session) {
  auto statement = connection_->prepareStatement(sqlSelectQuery_);

  auto rowset = statement->execute();

  int count = 0;
  size_t rowCount = 0;
  sql::JSONSQLWriter sqlWriter(isJSONPretty());
  sql::SQLRowsetProcessor sqlRowsetProcessor(rowset, { &sqlWriter });

  // Process rowset.
  do {
    rowCount = sqlRowsetProcessor.process(max_rows_ == 0 ? std::numeric_limits<size_t>::max() : max_rows_);
    count++;
    if (rowCount == 0)
      break;

    const auto output = sqlWriter.toString();
    if (!output.empty()) {
      WriteCallback writer(output);
      auto newflow = session.create();
      newflow->addAttribute(ResultRowCount, std::to_string(rowCount));
      session.write(newflow, &writer);
      session.transfer(newflow, s_success);
    }
  } while (rowCount > 0);
}

} /* namespace processors */
} /* namespace minifi */