/*
 * (c) Copyright, Real-Time Innovations, 2016-.
 * All rights reserved.
 * No duplications, whole or partial, manual or electronic, may be made
 * without express written permission.  Any such copies, or
 * revisions thereof, must display this notice unaltered.
 * This code contains trade secrets of Real-Time Innovations, Inc.
 */

#ifndef HPP_RTI_APPUTILS_SQLiteConnection_HPP_
#define HPP_RTI_APPUTILS_SQLiteConnection_HPP_

// IMPORTANT: macros.hpp must be the first RTI header included in every header
// file so that symbols are exported correctly on Windows
#include <dds/core/macros.hpp>

#include <sstream>
#include <iostream>
#include <string>
#include <vector>
#include <exception>
#include <chrono>
#include <thread>
#include <memory>
#include <algorithm>

#include "ndds/ndds_utility_c.h"
#include "dds_c/dds_c_log.h"

#include <dds/core/types.hpp>
#include <rti/core/constants.hpp>
#include <rti/core/Exception.hpp>
#include <rti/core/Semaphore.hpp>

#include "apputils/apputils_dll.h"

#include "dds/core/SafeEnumeration.hpp"
#include "dds/core/String.hpp"
#include "osapi/osapi_host.h"
#include "rtisqlite/sqlite3.h"
#include "rti/apputils/util/LockedBuffer.hpp"
#include "rti/apputils/util/Path.hpp"

struct sqlite3;


namespace rti { namespace apputils {

class APPUTILSDllExport SQLiteConnection {
public:
    class SQLiteException: public std::exception {
    public:

        SQLiteException() :
            sqlite_error_(),
            sqlite_error_code(SQLITE_OK),
            sql_statement_() {}

        SQLiteException(const SQLiteException& other) :
            sqlite_error_(other.sqlite_error_),
            sqlite_error_code(other.sqlite_error_code),
            sql_statement_(other.sql_statement_),
            what_(other.what_) {}

        SQLiteException(
                int return_code,
                const std::string error_string,
                const std::string sql_statement):
                sqlite_error_(error_string),
                sqlite_error_code(return_code),
                sql_statement_(sql_statement)
        {
            what_string_stream << "SQLite error: " + sqlite_error_;
            what_string_stream << std::endl;
            what_string_stream << "  [SQLite return code = " << sqlite_error_code
                    << "]" << std::endl;
            if (!sql_statement_.empty()) {
                what_string_stream << "  [SQL statement = " << sql_statement_
                        << "]" << std::endl;
            }
            set_what();
        }

        ~SQLiteException() throw() {}

        virtual const char * what() const throw()
        {
            return what_.c_str();
        }

    protected:
        void set_what() {
            size_t what_size = what_string_stream.str().size() + 1;
            what_.resize(what_size);
            strncpy(&what_[0], what_string_stream.str().c_str(), what_size);
        }

        std::string sqlite_error_;
        int sqlite_error_code;
        std::string sql_statement_;
        std::stringstream what_string_stream;
        std::string what_;
    };

    class SQLiteOpenException: public SQLiteException {
    public:
        SQLiteOpenException(
                const int return_code,
                const std::string error_string,
                const std::string filename)
                : filename_(filename)
        {
            sqlite_error_ = error_string;
            sqlite_error_code = return_code;

            what_string_stream << "SQLite - failed to open DB file; "
                    "sqlite returned error: "
                    + sqlite_error_;
            what_string_stream << std::endl;
            what_string_stream << "  [SQLite return code = "
                    << sqlite_error_code << "]" << std::endl;
            what_string_stream << "  [File = " << filename_ << "]"
                    << std::endl;
            set_what();
        }

        ~SQLiteOpenException() throw() {}

    private:
        std::string filename_;
    };

    class SQLiteCloseException: public SQLiteException {
    public:
        SQLiteCloseException(
                const int return_code,
                const std::string error_string)
        {
            sqlite_error_ = error_string;
            sqlite_error_code = return_code;

            what_string_stream
                    << "SQLite - failed to open DB; error: "
                    + sqlite_error_;
            what_string_stream << std::endl;
            what_string_stream << "  [SQLite return code = "
                    << sqlite_error_code << "]" << std::endl;
            set_what();
        }

        ~SQLiteCloseException() throw() {}

    };

    class SQLiteBindException: public SQLiteException {
    public:
        SQLiteBindException(
                int return_code,
                const std::string error_string,
                const std::string sql_statement,
                int column)
        {
            sqlite_error_ = error_string;
            sqlite_error_code = return_code;
            sql_statement_ = sql_statement;

            what_string_stream << "SQLite error: " + sqlite_error_;
            what_string_stream << std::endl;
            what_string_stream << "  [SQLite return code = "
                    << sqlite_error_code << "]" << std::endl;
            if (!sql_statement_.empty()) {
                what_string_stream << "  [SQL statement = " << sql_statement_
                        << "]" << std::endl;
            }
            what_string_stream << "  [Column being bound: " << column << "]"
                    << std::endl;
            set_what();
        };
    };

    static void delete_sqlite3(sqlite3 *sqlite)
    {
        if (sqlite != nullptr) {
            if (sqlite3_close_v2(sqlite) != SQLITE_OK) {
                const char *db_name = sqlite3_db_filename(sqlite, nullptr);
                DDSLogger_log(
                        RTI_LOG_BIT_EXCEPTION,
                        &RTI_LOG_FAILED_TO_CLOSE_FILE_TEMPLATE,
                        NULL, /* worker */
                        "Error closing SQLite DB; file name = %s",
                        (db_name != nullptr ? db_name : "<unknown>"));
            }
        }
    }

    static void delete_statement(sqlite3_stmt *statement)
    {
        if (statement != nullptr) {
            sqlite3_finalize(statement);
        }
    }

    /**
     * This class opens an exclusive SQLite transaction within the constructor,
     * on the SQLite connection given as a parameter to it.
     * The transaction is closed when the object comes out of scope, and the
     * destructor is called. It provides an error setting method. When there has
     * been an error, the transaction will be rolled back. When everything went
     * fine, the destructor will commit normally.
     */
    class TransactionToken {
    public:
        TransactionToken(SQLiteConnection& connection)
                : connection_(connection),
                  error_(false)
        {
            connection_.begin();
        }
        ~TransactionToken()
        {
            if (error_) {
                connection_.rollback();
            } else {
                connection_.commit();
            }
        }
        void set_error(bool error)
        {
            error_ = error;
        }
    private:
        SQLiteConnection& connection_;
        bool error_;
    };

    /**
     * @brief This class is a listener for the SQLiteConnection class. It is
     * used to notify when the associated connection is disconnected.
     * Notified objects should release all SQLite resources before returning.
     */
    class AbstractSQLiteDisconnectListener {
    public:
        /**
         * @brief Constructor
         *
         * @param connection The SQLite connection to listen to
         * @param ephimerous If true, this object will not be added to the
         * connection's list of listeners. This is useful for objects that
         * guarantee they will not outlive the connection. Default: false.
         */
        AbstractSQLiteDisconnectListener(
                SQLiteConnection& connection,
                bool ephemerous = true)
                : connection_(connection), ephemerous_(ephemerous)
        {
            if (!ephemerous_) {
                connection_.add_disconnect_listener(this);
            }
        }

        virtual ~AbstractSQLiteDisconnectListener()
        {
            if (!ephemerous_) {
                connection_.remove_disconnect_listener(this);
            }
        }

        virtual void on_disconnect() = 0;

    private:
        SQLiteConnection& connection_;
        bool ephemerous_;
    };

    class RowIterator : public AbstractSQLiteDisconnectListener {
    public:

        /**
         * @brief Constructor (from a statement).
         *
         * @param statement[in] The statement to iterate over.
         * @param connection[in] The connection to the database.
         * @param auto_step[in] If true, the iterator will step to the first row
         * when created. Default: true.
         * @param ephemerous[in] If true, this object will not be added to the
         * connection's list of listeners. This is useful for objects that
         * guarantee they will not outlive the connection. Default: true.
         *
         * @throws SQLiteException if there is an error resetting the statement.
         * @throws rti::core::PreconditionNotMetException if the statement is
         * invalid.
         */
        RowIterator(
                const std::shared_ptr<sqlite3_stmt>& statement,
                SQLiteConnection& connection,
                bool auto_step = true,
                bool ephemerous = true)
                : AbstractSQLiteDisconnectListener(connection, ephemerous),
                  statement_(statement),
                  connection_(connection),
                  return_code_(SQLITE_OK),
                  is_bound_(false),
                  is_output_stmt_(false)
        {
            initialize(auto_step);
        }

        /**
         * @brief Constructor (from SQL).
         *
         * @param statement[in] The SQL statement to iterate over.
         * @param connection[in] The connection to the database.
         * @param auto_step[in] If true, the iterator will step to the first row
         * when created. Default: true.
         * @param ephemerous[in] If true, this object will not be added to the
         * connection's list of listeners. This is useful for objects that
         * guarantee they will not outlive the connection. Default: true.
         *
         * @throws SQLiteException if there is an error preparing the statement.
         */
        RowIterator(
                const std::string& statement,
                SQLiteConnection& connection,
                bool auto_step = true,
                bool ephemerous = true)
                : AbstractSQLiteDisconnectListener(connection, ephemerous),
                  statement_(connection.prepare(statement)),
                  connection_(connection),
                  return_code_(SQLITE_OK),
                  is_bound_(false),
                  is_output_stmt_(false)
        {
            initialize(auto_step);
        }

        RowIterator& operator=(const RowIterator& other) = delete;

        virtual void on_disconnect() override
        {
            statement_ = nullptr;
        }

        bool is_valid() const
        {
            return statement_.get() != nullptr && connection_.connected();
        }

        int num_columns() const
        {
            RTI_CHECK_PRECONDITION(is_valid());
            return sqlite3_column_count(statement_.get());
        }

        void bind(const int column, const int64_t value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int ret_code = sqlite3_bind_int64(
                    statement_.get(),
                    column,
                    static_cast<sqlite_int64>(value));
            if (ret_code != SQLITE_OK) {
                std::string sql(sqlite3_sql(statement_.get()));
                throw SQLiteBindException(
                        ret_code,
                        connection_.last_err_msg(),
                        sql,
                        column);
            }
            is_bound_ = true;
        }

        void bind(const int column, const double value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int ret_code = sqlite3_bind_double(statement_.get(), column, value);
            if (ret_code != SQLITE_OK) {
                std::string sql(sqlite3_sql(statement_.get()));
                throw SQLiteBindException(
                        ret_code,
                        connection_.last_err_msg(),
                        sql,
                        column);
            }
            is_bound_ = true;
        }

        void bind(const int column, const std::string& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int ret_code = sqlite3_bind_text(
                    statement_.get(),
                    column,
                    value.c_str(),
                    static_cast<int>(value.size()),
                    SQLITE_TRANSIENT);
            if (ret_code != SQLITE_OK) {
                std::string sql(sqlite3_sql(statement_.get()));
                throw SQLiteBindException(
                        ret_code,
                        connection_.last_err_msg(),
                        sql,
                        column);
            }
            is_bound_ = true;
        }

        void bind(const int column, const std::vector<uint8_t>& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            void *value_array = nullptr;
            if (value.size() > 0) {
                value_array = (void *) (&(value[0]));
            }
            int ret_code = sqlite3_bind_blob(
                    statement_.get(),
                    column,
                    value_array,
                    static_cast<int>(value.size()),
                    SQLITE_TRANSIENT);
            if (ret_code != SQLITE_OK) {
                std::string sql(sqlite3_sql(statement_.get()));
                throw SQLiteBindException(
                        ret_code,
                        connection_.last_err_msg(),
                        sql,
                        column);
            }
            is_bound_ = true;
        }

        void bind(
                const int column,
                const rti::apputils::util::LockedBuffer<uint8_t>& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int ret_code = sqlite3_bind_blob(
                    statement_.get(),
                    column,
                    value.buffer(),
                    static_cast<int>(value.size()),
                    SQLITE_STATIC);
            if (ret_code != SQLITE_OK) {
                std::string sql(sqlite3_sql(statement_.get()));
                throw SQLiteBindException(
                        ret_code,
                        connection_.last_err_msg(),
                        sql,
                        column);
            }
            is_bound_ = true;
        }

        bool is_null(const int column)
        {
            RTI_CHECK_PRECONDITION(is_valid());
            return sqlite3_column_type(statement_.get(), column) == SQLITE_NULL;
        }

        /**
         * Get an integer value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(const int column, int64_t& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());
            value = static_cast<int64_t>(
                    sqlite3_column_int64(statement_.get(), column));
        }
        /**
         * Get a floating-point value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(const int column, double& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());
            value = sqlite3_column_double(statement_.get(), column);
        }
        /**
         * Get a text value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(const int column, std::string& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int length = sqlite3_column_bytes(statement_.get(), column);
            if (length > 0) {
                const char *c_tmp = (const char *)
                        sqlite3_column_text(statement_.get(), column);
                if (c_tmp == nullptr) {
                    std::string sql(sqlite3_sql(statement_.get()));
                    throw SQLiteException(
                            SQLITE_ERROR,
                            connection_.last_err_msg(),
                            sql);
                }
                value.assign(c_tmp, c_tmp + strlen(c_tmp));
            }
        }
        /**
         * Get a wide-string value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(const int column, dds::core::wstring& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int length = sqlite3_column_bytes(statement_.get(), column);
            if (length > 0) {
                DDS_UnsignedShort *utf16_string =
                        (DDS_UnsignedShort *) sqlite3_column_text16(
                                statement_.get(),
                                column);
                if (utf16_string == nullptr) {
                    std::string sql(sqlite3_sql(statement_.get()));
                    throw SQLiteException(
                            SQLITE_ERROR,
                            connection_.last_err_msg(),
                            sql);
                }
                /*
                 * The value returned by sqlite3_column_bytes16 cannot be
                 * negative
                 */
                uint32_t utf16_string_length =
                        (static_cast<uint32_t>(sqlite3_column_bytes16(
                                statement_.get(),
                                column)))
                        >> 1;
                dds::core::wstring tmp(utf16_string_length);
                for (uint32_t i = 0; i < utf16_string_length; ++i) {
                    tmp.native()[i] = static_cast<DDS_Wchar>(utf16_string[i]);
                }
                value = tmp;
            }
        }
        /**
         * Get a blob value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(const int column, std::vector<uint8_t>& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int length = sqlite3_column_bytes(statement_.get(), column);
            if (length > 0) {
                uint8_t *c_array = (uint8_t *)
                        sqlite3_column_blob(statement_.get(), column);
                value.assign(c_array, c_array + length);
            }
        }
        /**
         * Get a blob pointer value from the current iterator position.
         *
         * @param column The index of the column to get the value from. Leftmost
         *               column is indexed by 0.
         */
        void get(
                const int column,
                rti::apputils::util::LockedBuffer<uint8_t>& value)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            int length = sqlite3_column_bytes(statement_.get(), column);
            if (length > 0) {
                uint8_t *c_array = (uint8_t *)
                        sqlite3_column_blob(statement_.get(), column);
                /* We know length is positive */
                value.set(c_array, static_cast<uint32_t>(length));
            }
        }

        bool has_no_data(const int column)
        {
            RTI_CHECK_PRECONDITION(is_valid());
            return (sqlite3_column_bytes(statement_.get(), column) == 0);
        }

        /**
         * Step the iterator.
         */
        RowIterator& operator ++()
        {
            RTI_CHECK_PRECONDITION(is_valid());

            return_code_ = sqlite3_step(statement_.get());
            switch (return_code_) {
            case (SQLITE_OK):
                break;
            case (SQLITE_ROW):
                break;
            case (SQLITE_DONE):
                break;
            default:
                throw SQLiteException(
                        return_code_,
                        connection_.last_err_msg(),
                        sqlite3_sql(statement_.get()));
            }
            if (is_bound_ && is_output_stmt_) {
                int ret_code = sqlite3_reset(statement_.get());
                if (ret_code != SQLITE_OK) {
                    throw SQLiteException(
                            ret_code,
                            connection_.last_err_msg(),
                            sqlite3_sql(statement_.get()));
                }
                sqlite3_clear_bindings(statement_.get());
            }
            return *this;
        }

        bool at_end() {
            return return_code_ == SQLITE_DONE;
        }

        void reset() {
            RTI_CHECK_PRECONDITION(is_valid());

            int ret_code = sqlite3_reset(statement_.get());
            if (ret_code != SQLITE_OK) {
                throw SQLiteException(
                        ret_code,
                        connection_.last_err_msg(),
                        sqlite3_sql(statement_.get()));
            }
            sqlite3_clear_bindings(statement_.get());
            return_code_ = SQLITE_OK;
        }

    private:
        std::shared_ptr<sqlite3_stmt> statement_;
        SQLiteConnection& connection_;
        int return_code_;
        bool is_bound_;

        /**
         * This is true only when the statement is an INSERT statement
         */
        bool is_output_stmt_;

        void initialize(const bool auto_step)
        {
            RTI_CHECK_PRECONDITION(is_valid());

            return_code_ = sqlite3_reset(statement_.get());
            if (return_code_ != SQLITE_OK) {
                throw SQLiteException(
                        return_code_,
                        connection_.last_err_msg(),
                        sqlite3_sql(statement_.get()));
            }
            if (REDAString_iFindSubString(
                        sqlite3_sql(statement_.get()),
                        "INSERT INTO ") == nullptr
                    && REDAString_iFindSubString(
                           sqlite3_sql(statement_.get()),
                           "DELETE FROM ") == nullptr) {
                if (auto_step) {
                    ++(*this);
                }
            } else {
                is_output_stmt_ = true;
            }
        }
    };

    SQLiteConnection()
            : connection_(nullptr, delete_sqlite3),
              connected_(false),
              sqlite_init_string_(default_init_string)
    { }

    SQLiteConnection(const std::string& sqlite_init_string)
            : connection_(nullptr, delete_sqlite3),
              connected_(false),
              sqlite_init_string_(sqlite_init_string)
    { }

    SQLiteConnection(const SQLiteConnection& other) = delete;

    SQLiteConnection(SQLiteConnection&& other)
            : connection_(nullptr, delete_sqlite3),
              connected_(other.connected_),
              sqlite_init_string_(other.sqlite_init_string_)
    {
        connection_.reset(other.connection_.release());
    }

    ~SQLiteConnection()
    {
        if (connected_) {
            disconnect();
        }
    }

    SQLiteConnection& operator=(SQLiteConnection& other) = delete;

    SQLiteConnection& operator=(SQLiteConnection&& other)
    {
        connection_.reset(other.connection_.release());
        sqlite_init_string_ = other.sqlite_init_string_;
        return *this;
    }

    void connect(
            const std::string& filename,
            bool create_on_open = true,
            bool readonly = false)
    {
        struct sqlite3 *connection = nullptr;
        std::string sql;

        if (create_on_open && readonly) {
            throw std::runtime_error(
                    "precondition not met: not (create_on_open and readonly)");
        }

        // Setup the SQLite open function flags
        int open_flags = 0;
        std::string stringified_open_flags(" for ");
        if (create_on_open) {
            /*
             * Create the file when opening it. Replay and Converter _should not_
             * use this
             */
            open_flags = open_flags | SQLITE_OPEN_CREATE;
            stringified_open_flags += "create ";
        }
        if (readonly) {
            open_flags = open_flags | SQLITE_OPEN_READONLY;
            stringified_open_flags += "readonly ";
        } else {
            open_flags = open_flags | SQLITE_OPEN_READWRITE;
            stringified_open_flags += "read+write ";
        }
        int sqlite3_open_retcode = sqlite3_open_v2(
                filename.c_str(),
                &connection,
                open_flags,
                nullptr);
        if (sqlite3_open_retcode != SQLITE_OK) {
            std::stringstream location_to_report;
#if !defined(RTI_INTY)
            location_to_report << filename << " from working directory ";
            location_to_report << rti::apputils::util::Path::cwd();
#else
            location_to_report << filename << " ";
#endif
            location_to_report << stringified_open_flags;
            sqlite3_close_v2(connection);
            throw SQLiteOpenException(
                    sqlite3_open_retcode,
                    sqlite3_errstr(sqlite3_open_retcode),
                    location_to_report.str());
        }
        connection_.reset(connection);

        if (sqlite3_busy_handler(
                connection,
                SQLiteConnection::busy_handler,
                nullptr) != SQLITE_OK) {
            std::stringstream location_to_report;
#if !defined(RTI_INTY)
            location_to_report << filename << " from working directory ";
            location_to_report << rti::apputils::util::Path::cwd();
#else
            location_to_report << filename << " ";
#endif
            location_to_report << stringified_open_flags;
            throw SQLiteOpenException(
                    sqlite3_errcode(connection_.get()),
                    sqlite3_errmsg(connection_.get()),
                    location_to_report.str());
        }

        // RECNEXTGEN-72 [Juanlu]: for the EAR 1, default to the most performant
        // SQLite settings. We will need to document that we're setting them
        exec(sqlite_init_string_);

        connected_ = true;
    }

    bool connected() const
    {
        return connected_;
    }

    const std::string db_filename() const
    {
        const char *sqlite_file =
                sqlite3_db_filename(connection_.get(), nullptr);
        if (sqlite_file != nullptr) {
            return std::string(sqlite_file);
        }
        return std::string();
    }

    void add_disconnect_listener(AbstractSQLiteDisconnectListener *listener)
    {
        disconnect_listeners_.push_back(listener);
    }

    void remove_disconnect_listener(AbstractSQLiteDisconnectListener *listener)
    {
#if defined(__cplusplus) && __cplusplus >= 201103L

        const auto it = std::find(
                disconnect_listeners_.begin(),
                disconnect_listeners_.end(),
                listener);
        if (it == disconnect_listeners_.end()) {
            DDSLogger_log(
                    RTI_LOG_BIT_WARN,
                    &RTI_LOG_FAILED_TO_FIND_TEMPLATE,
                    NULL, /* worker */
                    "disconnect listener");
            return;
        }
        disconnect_listeners_.erase(it);
#else
        /*
         * Web Integration Service uses SQLiteConnection but is written in pre
         * C++11 standard
         */
        for (std::vector<AbstractSQLiteDisconnectListener *>::iterator it =
                     disconnect_listeners_.begin();
             it != disconnect_listeners_.end();
             ++it) {
            if (*it == listener) {
                disconnect_listeners_.erase(it);
                break;
            }
        }
#endif
    }

    void disconnect()
    {
        for (AbstractSQLiteDisconnectListener *listener :
             disconnect_listeners_) {
            listener->on_disconnect();
        }
        connection_.reset(nullptr);
        connected_ = false;
    }

    void begin()
    {
        std::string sql("BEGIN");
        int retcode = sqlite3_exec(
                connection_.get(),
                sql.c_str(),
                nullptr,
                nullptr,
                nullptr);
        if (retcode != SQLITE_OK && retcode != SQLITE_BUSY) {
            throw SQLiteException(
                    sqlite3_errcode(connection_.get()),
                    sqlite3_errmsg(connection_.get()),
                    sql);
        }
        while (retcode == SQLITE_BUSY) {
            retcode = sqlite3_exec(
                    connection_.get(),
                    sql.c_str(),
                    nullptr,
                    nullptr,
                    nullptr);
            if (retcode != SQLITE_OK && retcode != SQLITE_BUSY) {
                throw SQLiteException(
                        sqlite3_errcode(connection_.get()),
                        sqlite3_errmsg(connection_.get()),
                        sql);
            }
        }
    }

    void commit()
    {
        const std::string sql("COMMIT");

        if (sqlite3_exec(
                    connection_.get(),
                    sql.c_str(),
                    nullptr,
                    nullptr,
                    nullptr) != SQLITE_OK) {
            throw SQLiteException(
                    sqlite3_errcode(connection_.get()),
                    sqlite3_errmsg(connection_.get()),
                    sql);
        }
    }

    void rollback()
    {
        const std::string sql("ROLLBACK");

        if (sqlite3_exec(
                    connection_.get(),
                    sql.c_str(),
                    nullptr,
                    nullptr,
                    nullptr) != SQLITE_OK) {
            throw SQLiteException(
                    sqlite3_errcode(connection_.get()),
                    sqlite3_errmsg(connection_.get()),
                    sql);
        }
    }

    std::shared_ptr<sqlite3_stmt> prepare(const std::string& statement_text)
    {
        sqlite3_stmt *statement = nullptr;

        if (sqlite3_prepare_v2(
                    connection_.get(),
                    statement_text.c_str(),
                    (int) statement_text.size(),
                    &statement,
                    nullptr) != SQLITE_OK) {
            throw SQLiteException(
                    sqlite3_errcode(connection_.get()),
                    sqlite3_errmsg(connection_.get()),
                    statement_text);
        }
        return std::shared_ptr<sqlite3_stmt>(statement, delete_statement);
    }

    void exec(const std::string& statement_text)
    {
        int return_code = sqlite3_exec(
                connection_.get(),
                statement_text.c_str(),
                nullptr,
                nullptr,
                nullptr);
        if (return_code != SQLITE_OK) {
            throw SQLiteException(
                    return_code,
                    sqlite3_errmsg(connection_.get()),
                    statement_text);
        }
    }

    const std::string last_err_msg() const
    {
        return sqlite3_errmsg(connection_.get());
    }

    RowIterator begin(
            const std::shared_ptr<sqlite3_stmt>& statement,
            bool auto_step = true,
            bool ephemerous = true)
    {
        return RowIterator(statement, *this, auto_step, ephemerous);
    }

    RowIterator begin(
            const std::string& statement_text,
            bool auto_step = true,
            bool ephemerous = true)
    {
        return RowIterator(statement_text, *this, auto_step, ephemerous);
    }

    const std::string& sqlite_init_string() { return sqlite_init_string_; }

    /**
     * See: https://www.sqlite.org/c3ref/busy_handler.html
     * The SQLite busy handler allows for a process to wait when another process
     * has a database table locked. This will help concurrency, but also, is
     * necessary for some of our automated tests.
     */
    static int busy_handler(void *, int)
    {
        const int32_t max_time_millis = 300;
        const auto timeout = std::chrono::milliseconds(
                RTIOsapiUtility_randRange(0, max_time_millis));

        const std::string timeout_str =
                std::to_string(timeout.count()) + " milliseconds";
        DDSLogger_log(
                RTI_LOG_BIT_CONTENT,
                &RTI_LOG_EMPTY_TEMPLATE,
                NULL,
                "Executing SQLite busy handler: waiting for %s",
                timeout_str.c_str());

        std::this_thread::sleep_for(timeout);
        return RTI_TRUE;
    }

    static constexpr const char * const default_init_string {
        "PRAGMA SYNCHRONOUS = OFF; PRAGMA JOURNAL_MODE = MEMORY;"
    };

    bool table_exists(const std::string& table_name)
    {
        try {
            std::stringstream select_stmt_stream;
            select_stmt_stream
                    << "SELECT * FROM [" << table_name << "] LIMIT 1";
            exec(select_stmt_stream.str());
        } catch (...) {
            return false;
        }
        return true;
    }

    int64_t last_rowid() const
    {
        return sqlite3_last_insert_rowid(connection_.get());
    }

private:
    std::unique_ptr<sqlite3, decltype(&delete_sqlite3)> connection_;
    bool connected_;
    std::vector<AbstractSQLiteDisconnectListener *> disconnect_listeners_;
    std::string sqlite_init_string_;
};

/**
 * This class is supposed to be sub-classed by classes that contain an
 * SQLite connection, for example, our metadata or discovery file
 * abstractions.
 */
class AbstractSQLiteObject {
public:
    AbstractSQLiteObject(
            const std::string& sqlite_init_string =
                    SQLiteConnection::default_init_string)
            : connection_(sqlite_init_string)
    { }

    virtual ~AbstractSQLiteObject()
    { }

    virtual SQLiteConnection& connection() { return connection_; }

protected:

    SQLiteConnection connection_;
};


} } /* rti::app_utils */

#endif /* HPP_RTI_APPUTILS_SQLiteConnection_HPP_ */