DatabaseTopicReader.cpp

Go to the documentation of this file.

/*
* This file is part of ArmarX.
*
* Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
*
* ArmarX is free software; you can redistribute it and/or modify
* it under the terms of the GNU General Public License version 2 as
* published by the Free Software Foundation.
*
* ArmarX is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*
* @package    ArmarX
* @author     Philipp Schmidt( ufedv at student dot kit dot edu)
* @date       2016
* @copyright  http://www.gnu.org/licenses/gpl-2.0.txt
*             GNU General Public License
*/
#include "DatabaseTopicReader.h"
 
#include "ArmarXCore/core/logging/Logging.h"
 
namespace armarx
{
 
    DatabaseTopicReader::DatabaseTopicReader(std::filesystem::path path) :
        filepath(path), db(nullptr), stmt(nullptr), database_open(true), end_of_database(false)
    {
        //Open database
        int error_code = sqlite3_open(filepath.c_str(), &db);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR_S << "Error opening database: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "Database Path: " << filepath.c_str();
            sqlite3_close(db);
            database_open = false;
            return;
        }
 
        //Prepare SQL statement
        std::string sql_select = "SELECT * FROM TopicData WHERE ID >= ?";
        error_code = sqlite3_prepare_v2(db, sql_select.c_str(), -1, &stmt, nullptr);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_select;
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            database_open = false;
            return;
        }
 
        //Seek to start position
        if (!seekTo(1))
        {
            ARMARX_ERROR_S << "Can not seek to start position, or maybe database is empty.";
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            database_open = false;
            return;
        }
 
        //Check if layout is correct (at least correct amount of columns)
        if (sqlite3_column_count(stmt) != 5)
        {
            ARMARX_ERROR_S << "Database table layout is wrong. Closing database.";
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            database_open = false;
            return;
        }
    }
 
    DatabaseTopicReader::~DatabaseTopicReader()
    {
        //NULL pointer is a harmless op
        sqlite3_finalize(stmt);
        sqlite3_close(db);
    }
 
    std::filesystem::path
    DatabaseTopicReader::getFilepath() const
    {
        return filepath;
    }
 
    bool
    DatabaseTopicReader::read(TopicUtil::TopicData& data)
    {
        //Make sure this is threadsafe
        std::unique_lock lock(mutex);
 
        //Check if database is open
        if (!database_open)
        {
            ARMARX_ERROR_S << "Database not open or database table layout is wrong";
            return false;
        }
 
        //Check if end of select statement is reached
        if (end_of_database)
        {
            //ARMARX_ERROR << "Can not read, end of replay data reached";
            return false;
        }
 
 
        //Return current row
        data.timestamp = IceUtil::Time::microSeconds(sqlite3_column_int64(stmt, 1));
        data.topicName = std::string(reinterpret_cast<const char*>(sqlite3_column_text(stmt, 2)));
        data.operationName =
            std::string(reinterpret_cast<const char*>(sqlite3_column_text(stmt, 3)));
 
        //Retreive blob data and store in vector
        const void* blob_data = sqlite3_column_blob(stmt, 4);
        int size = sqlite3_column_bytes(stmt, 4);
        //vector range constructor should copy the data according to docu
        data.inParams = std::vector<Ice::Byte>((Ice::Byte*)blob_data, (Ice::Byte*)blob_data + size);
 
        //Move to next row for next read access
        int error_code = sqlite3_step(stmt);
        //We might get another row or the end of the list
        if (error_code != SQLITE_ROW)
        {
            //Maybe end of select statement?
            if (error_code == SQLITE_DONE)
            {
                ARMARX_INFO << "End of replay reached";
                end_of_database = true;
            }
            else
            {
                ARMARX_ERROR_S << "Retrieving next row of data failed, closing database: "
                               << sqlite3_errmsg(db);
                sqlite3_finalize(stmt);
                sqlite3_close(db);
                database_open = false;
                return false;
            }
        }
 
        return true;
    }
 
    bool
    DatabaseTopicReader::seekTo(IceUtil::Time timestamp)
    {
        sqlite3_stmt* stmt_get_id;
        std::string sql_get_id = "SELECT ID FROM TopicData WHERE TIME >= ? LIMIT 1";
 
        int error_code = sqlite3_prepare_v2(db, sql_get_id.c_str(), -1, &stmt_get_id, nullptr);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
            ARMARX_ERROR << "Can not prepare sql statement: " << sqlite3_errmsg(db);
            ARMARX_ERROR << "SQL Query: " << sql_get_id;
            sqlite3_finalize(stmt_get_id);
            return false;
        }
 
        error_code = sqlite3_bind_int64(stmt_get_id, 1, timestamp.toMicroSeconds());
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
            ARMARX_ERROR << "Can not bind parameter to sql query: " << sqlite3_errmsg(db);
            ARMARX_ERROR << "SQL Query: " << sql_get_id;
            sqlite3_finalize(stmt_get_id);
            return false;
        }
 
        error_code = sqlite3_step(stmt_get_id);
        //Expecting some results, otherwise something went wrong
        if (error_code != SQLITE_ROW)
        {
            ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
            ARMARX_ERROR << "Query did not succeed or we got no result: " << sqlite3_errmsg(db);
            ARMARX_ERROR << "SQL Query: " << sql_get_id;
            sqlite3_finalize(stmt_get_id);
            return false;
        }
 
        bool returnValue = seekTo(sqlite3_column_int(stmt_get_id, 0));
        sqlite3_finalize(stmt_get_id);
        return returnValue;
    }
 
    IceUtil::Time
    DatabaseTopicReader::getReplayLength()
    {
        // Get timestamp of last entry
        sqlite3_stmt* stmt_replay_length;
        std::string sql_replay_length = "SELECT max(TIME) FROM TopicData";
        int error_code =
            sqlite3_prepare_v2(db, sql_replay_length.c_str(), -1, &stmt_replay_length, nullptr);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_replay_length;
            sqlite3_finalize(stmt_replay_length);
            return IceUtil::Time::microSeconds(0);
        }
        error_code = sqlite3_step(stmt_replay_length);
        if (error_code != SQLITE_ROW)
        {
            ARMARX_ERROR_S << "Can not execute sql statement, or no result: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_replay_length;
            sqlite3_finalize(stmt_replay_length);
            return IceUtil::Time::microSeconds(0);
        }
 
        long returnValue = sqlite3_column_int64(stmt_replay_length, 0);
        sqlite3_finalize(stmt_replay_length);
        return IceUtil::Time::microSeconds(returnValue);
    }
 
    std::vector<std::string>
    DatabaseTopicReader::getReplayTopics()
    {
        std::vector<std::string> topicList;
 
        //Check if database is open
        if (!database_open)
        {
            ARMARX_ERROR << "Can not read list of replay topics if database is not open";
            return topicList;
        }
 
        //Retrieve all topics of database
        sqlite3_stmt* stmt_topic_list;
        std::string sql_topic_list = "SELECT TOPIC FROM Topics";
        int error_code =
            sqlite3_prepare_v2(db, sql_topic_list.c_str(), -1, &stmt_topic_list, nullptr);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
            sqlite3_finalize(stmt_topic_list);
            return topicList;
        }
        error_code = sqlite3_step(stmt_topic_list);
        if (error_code != SQLITE_ROW)
        {
            ARMARX_ERROR_S << "Can not execute sql statement, or no result: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
            sqlite3_finalize(stmt_topic_list);
            return topicList;
        }
 
        //Process until end of list is reached
        while (error_code == SQLITE_ROW)
        {
            topicList.push_back(std::string(
                reinterpret_cast<const char*>(sqlite3_column_text(stmt_topic_list, 0))));
            error_code = sqlite3_step(stmt_topic_list);
        }
 
        //Just for good practice
        //State after reading all columns should be SQLITE_DONE
        if (error_code != SQLITE_DONE)
        {
            ARMARX_ERROR_S << "Error reading topic list: " << sqlite3_errmsg(db);
            ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
            sqlite3_finalize(stmt_topic_list);
            return topicList;
        }
 
        sqlite3_finalize(stmt_topic_list);
        return topicList;
    }
 
    bool
    DatabaseTopicReader::seekTo(int ID)
    {
        //Make sure this is threadsafe
        std::unique_lock lock(mutex);
 
        //Check if database is open
        if (!database_open)
        {
            ARMARX_ERROR << "Can not seek to position if database is not open";
            return false;
        }
 
        //Reset statement
        sqlite3_reset(stmt);
        sqlite3_clear_bindings(stmt);
 
        //Reset end of database
        end_of_database = false;
 
        //Bind ID
        int error_code = sqlite3_bind_int(stmt, 1, ID);
        if (error_code != SQLITE_OK)
        {
            ARMARX_ERROR_S << "Could not bind data while trying to seek: " << sqlite3_errmsg(db);
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            database_open = false;
            return false;
        }
 
        error_code = sqlite3_step(stmt);
        //Expecting some results, otherwise we are at the end of database or something went wrong
        if (error_code != SQLITE_ROW)
        {
            ARMARX_ERROR_S << "Query did not succeed or we got no result while trying to seek: "
                           << sqlite3_errmsg(db);
            sqlite3_finalize(stmt);
            sqlite3_close(db);
            database_open = false;
            return false;
        }
 
        return true;
    }
 
} // namespace armarx