Method onTrigger

extensions/sqlite/ExecuteSQL.cpp:69–173

}

void ExecuteSQL::onTrigger(const std::shared_ptr<core::ProcessContext> &context,
                           const std::shared_ptr<core::ProcessSession> &session) {
  auto flow_file = session->get();

  try {
    // Use an existing context, if one is available
    std::shared_ptr<minifi::sqlite::SQLiteConnection> db;

    if (conn_q_.try_dequeue(db)) {
      logger_->log_debug("Using available SQLite connection");
    }

    if (!db) {
      logger_->log_info("Creating new SQLite connection");
      if (db_url_.substr(0, 9) == "sqlite://") {
        db = std::make_shared<minifi::sqlite::SQLiteConnection>(db_url_.substr(9));
      } else {
        std::stringstream err_msg;
        err_msg << "Connection URL '" << db_url_ << "' is unsupported";
        logger_->log_error(err_msg.str().c_str());
        throw std::runtime_error("Connection Error");
      }
    }

    auto dynamic_sql = std::make_shared<std::string>();

    if (flow_file) {
      if (sql_.empty()) {
        // SQL is not defined as a property, so get SQL from the file content
        SQLReadCallback cb(dynamic_sql);
        session->read(flow_file, &cb);
      } else {
        // SQL is defined as a property, so get the property dynamically w/ EL support
        context->getProperty(SQLStatement, *dynamic_sql, flow_file);
      }
    }

    auto stmt = flow_file
                ? db->prepare(*dynamic_sql)
                : db->prepare(sql_);

    if (flow_file) {
      for (int i = 1; i < INT_MAX; i++) {
        std::string val;
        std::stringstream val_key;
        val_key << "sql.args." << i << ".value";

        if (!flow_file->getAttribute(val_key.str(), val)) {
          break;
        }

        stmt.bind_text(i, val);
      }
    }

    stmt.step();

    if (!stmt.is_ok()) {