Method onTrigger

extensions/sqlite/PutSQL.cpp:83–173

}

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

  if (!flow_file) {
    return;
  }

  uint64_t batch_processed = 1;

  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");
      }
    }

    do {
      auto sql = std::make_shared<std::string>();

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

      auto stmt = db->prepare(*sql);
      for (uint64_t i = 1; i < UINT64_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()) {