DatabaseTopicReader.cpp
Go to the documentation of this file.
1 /*
2 * This file is part of ArmarX.
3 *
4 * Copyright (C) 2011-2016, High Performance Humanoid Technologies (H2T), Karlsruhe Institute of Technology (KIT), all rights reserved.
5 *
6 * ArmarX is free software; you can redistribute it and/or modify
7 * it under the terms of the GNU General Public License version 2 as
8 * published by the Free Software Foundation.
9 *
10 * ArmarX is distributed in the hope that it will be useful, but
11 * WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
14 *
15 * You should have received a copy of the GNU General Public License
16 * along with this program. If not, see <http://www.gnu.org/licenses/>.
17 *
18 * @package ArmarX
19 * @author Philipp Schmidt( ufedv at student dot kit dot edu)
20 * @date 2016
21 * @copyright http://www.gnu.org/licenses/gpl-2.0.txt
22 * GNU General Public License
23 */
24 #include "DatabaseTopicReader.h"
25 #include "ArmarXCore/core/logging/Logging.h"
26 
27 namespace armarx
28 {
29 
30  DatabaseTopicReader::DatabaseTopicReader(std::filesystem::path path) :
31  filepath(path), db(nullptr), stmt(nullptr), database_open(true), end_of_database(false)
32  {
33  //Open database
34  int error_code = sqlite3_open(filepath.c_str(), &db);
35  if (error_code != SQLITE_OK)
36  {
37  ARMARX_ERROR_S << "Error opening database: " << sqlite3_errmsg(db);
38  ARMARX_ERROR_S << "Database Path: " << filepath.c_str();
39  sqlite3_close(db);
40  database_open = false;
41  return;
42  }
43 
44  //Prepare SQL statement
45  std::string sql_select = "SELECT * FROM TopicData WHERE ID >= ?";
46  error_code = sqlite3_prepare_v2(db, sql_select.c_str(), -1, &stmt, nullptr);
47  if (error_code != SQLITE_OK)
48  {
49  ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
50  ARMARX_ERROR_S << "SQL Query: " << sql_select;
51  sqlite3_finalize(stmt);
52  sqlite3_close(db);
53  database_open = false;
54  return;
55  }
56 
57  //Seek to start position
58  if (!seekTo(1))
59  {
60  ARMARX_ERROR_S << "Can not seek to start position, or maybe database is empty.";
61  sqlite3_finalize(stmt);
62  sqlite3_close(db);
63  database_open = false;
64  return;
65  }
66 
67  //Check if layout is correct (at least correct amount of columns)
68  if (sqlite3_column_count(stmt) != 5)
69  {
70  ARMARX_ERROR_S << "Database table layout is wrong. Closing database.";
71  sqlite3_finalize(stmt);
72  sqlite3_close(db);
73  database_open = false;
74  return;
75  }
76  }
77 
78  DatabaseTopicReader::~DatabaseTopicReader()
79  {
80  //NULL pointer is a harmless op
81  sqlite3_finalize(stmt);
82  sqlite3_close(db);
83  }
84 
85  std::filesystem::path DatabaseTopicReader::getFilepath() const
86  {
87  return filepath;
88  }
89 
90  bool DatabaseTopicReader::read(TopicUtil::TopicData& data)
91  {
92  //Make sure this is threadsafe
93  std::unique_lock lock(mutex);
94 
95  //Check if database is open
96  if (!database_open)
97  {
98  ARMARX_ERROR_S << "Database not open or database table layout is wrong";
99  return false;
100  }
101 
102  //Check if end of select statement is reached
103  if (end_of_database)
104  {
105  //ARMARX_ERROR << "Can not read, end of replay data reached";
106  return false;
107  }
108 
109 
110  //Return current row
111  data.timestamp = IceUtil::Time::microSeconds(sqlite3_column_int64(stmt, 1));
112  data.topicName = std::string(reinterpret_cast<const char*>(sqlite3_column_text(stmt, 2)));
113  data.operationName = std::string(reinterpret_cast<const char*>(sqlite3_column_text(stmt, 3)));
114 
115  //Retreive blob data and store in vector
116  const void* blob_data = sqlite3_column_blob(stmt, 4);
117  int size = sqlite3_column_bytes(stmt, 4);
118  //vector range constructor should copy the data according to docu
119  data.inParams = std::vector<Ice::Byte>((Ice::Byte*)blob_data, (Ice::Byte*)blob_data + size);
120 
121  //Move to next row for next read access
122  int error_code = sqlite3_step(stmt);
123  //We might get another row or the end of the list
124  if (error_code != SQLITE_ROW)
125  {
126  //Maybe end of select statement?
127  if (error_code == SQLITE_DONE)
128  {
129  ARMARX_INFO << "End of replay reached";
130  end_of_database = true;
131  }
132  else
133  {
134  ARMARX_ERROR_S << "Retrieving next row of data failed, closing database: " << sqlite3_errmsg(db);
135  sqlite3_finalize(stmt);
136  sqlite3_close(db);
137  database_open = false;
138  return false;
139  }
140  }
141 
142  return true;
143  }
144 
145  bool DatabaseTopicReader::seekTo(IceUtil::Time timestamp)
146  {
147  sqlite3_stmt* stmt_get_id;
148  std::string sql_get_id = "SELECT ID FROM TopicData WHERE TIME >= ? LIMIT 1";
149 
150  int error_code = sqlite3_prepare_v2(db, sql_get_id.c_str(), -1, &stmt_get_id, nullptr);
151  if (error_code != SQLITE_OK)
152  {
153  ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
154  ARMARX_ERROR << "Can not prepare sql statement: " << sqlite3_errmsg(db);
155  ARMARX_ERROR << "SQL Query: " << sql_get_id;
156  sqlite3_finalize(stmt_get_id);
157  return false;
158  }
159 
160  error_code = sqlite3_bind_int64(stmt_get_id, 1, timestamp.toMicroSeconds());
161  if (error_code != SQLITE_OK)
162  {
163  ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
164  ARMARX_ERROR << "Can not bind parameter to sql query: " << sqlite3_errmsg(db);
165  ARMARX_ERROR << "SQL Query: " << sql_get_id;
166  sqlite3_finalize(stmt_get_id);
167  return false;
168  }
169 
170  error_code = sqlite3_step(stmt_get_id);
171  //Expecting some results, otherwise something went wrong
172  if (error_code != SQLITE_ROW)
173  {
174  ARMARX_ERROR << "Could not seek to timestamp: " << timestamp.toMicroSeconds();
175  ARMARX_ERROR << "Query did not succeed or we got no result: " << sqlite3_errmsg(db);
176  ARMARX_ERROR << "SQL Query: " << sql_get_id;
177  sqlite3_finalize(stmt_get_id);
178  return false;
179  }
180 
181  bool returnValue = seekTo(sqlite3_column_int(stmt_get_id, 0));
182  sqlite3_finalize(stmt_get_id);
183  return returnValue;
184  }
185 
186  IceUtil::Time DatabaseTopicReader::getReplayLength()
187  {
188  // Get timestamp of last entry
189  sqlite3_stmt* stmt_replay_length;
190  std::string sql_replay_length = "SELECT max(TIME) FROM TopicData";
191  int error_code = sqlite3_prepare_v2(db, sql_replay_length.c_str(), -1, &stmt_replay_length, nullptr);
192  if (error_code != SQLITE_OK)
193  {
194  ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
195  ARMARX_ERROR_S << "SQL Query: " << sql_replay_length;
196  sqlite3_finalize(stmt_replay_length);
197  return IceUtil::Time::microSeconds(0);
198  }
199  error_code = sqlite3_step(stmt_replay_length);
200  if (error_code != SQLITE_ROW)
201  {
202  ARMARX_ERROR_S << "Can not execute sql statement, or no result: " << sqlite3_errmsg(db);
203  ARMARX_ERROR_S << "SQL Query: " << sql_replay_length;
204  sqlite3_finalize(stmt_replay_length);
205  return IceUtil::Time::microSeconds(0);
206  }
207 
208  long returnValue = sqlite3_column_int64(stmt_replay_length, 0);
209  sqlite3_finalize(stmt_replay_length);
210  return IceUtil::Time::microSeconds(returnValue);
211  }
212 
213  std::vector<std::string> DatabaseTopicReader::getReplayTopics()
214  {
215  std::vector<std::string> topicList;
216 
217  //Check if database is open
218  if (!database_open)
219  {
220  ARMARX_ERROR << "Can not read list of replay topics if database is not open";
221  return topicList;
222  }
223 
224  //Retrieve all topics of database
225  sqlite3_stmt* stmt_topic_list;
226  std::string sql_topic_list = "SELECT TOPIC FROM Topics";
227  int error_code = sqlite3_prepare_v2(db, sql_topic_list.c_str(), -1, &stmt_topic_list, nullptr);
228  if (error_code != SQLITE_OK)
229  {
230  ARMARX_ERROR_S << "Can not prepare sql statement: " << sqlite3_errmsg(db);
231  ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
232  sqlite3_finalize(stmt_topic_list);
233  return topicList;
234  }
235  error_code = sqlite3_step(stmt_topic_list);
236  if (error_code != SQLITE_ROW)
237  {
238  ARMARX_ERROR_S << "Can not execute sql statement, or no result: " << sqlite3_errmsg(db);
239  ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
240  sqlite3_finalize(stmt_topic_list);
241  return topicList;
242  }
243 
244  //Process until end of list is reached
245  while (error_code == SQLITE_ROW)
246  {
247  topicList.push_back(std::string(reinterpret_cast<const char*>(sqlite3_column_text(stmt_topic_list, 0))));
248  error_code = sqlite3_step(stmt_topic_list);
249  }
250 
251  //Just for good practice
252  //State after reading all columns should be SQLITE_DONE
253  if (error_code != SQLITE_DONE)
254  {
255  ARMARX_ERROR_S << "Error reading topic list: " << sqlite3_errmsg(db);
256  ARMARX_ERROR_S << "SQL Query: " << sql_topic_list;
257  sqlite3_finalize(stmt_topic_list);
258  return topicList;
259  }
260 
261  sqlite3_finalize(stmt_topic_list);
262  return topicList;
263  }
264 
265  bool DatabaseTopicReader::seekTo(int ID)
266  {
267  //Make sure this is threadsafe
268  std::unique_lock lock(mutex);
269 
270  //Check if database is open
271  if (!database_open)
272  {
273  ARMARX_ERROR << "Can not seek to position if database is not open";
274  return false;
275  }
276 
277  //Reset statement
278  sqlite3_reset(stmt);
279  sqlite3_clear_bindings(stmt);
280 
281  //Reset end of database
282  end_of_database = false;
283 
284  //Bind ID
285  int error_code = sqlite3_bind_int(stmt, 1, ID);
286  if (error_code != SQLITE_OK)
287  {
288  ARMARX_ERROR_S << "Could not bind data while trying to seek: " << sqlite3_errmsg(db);
289  sqlite3_finalize(stmt);
290  sqlite3_close(db);
291  database_open = false;
292  return false;
293  }
294 
295  error_code = sqlite3_step(stmt);
296  //Expecting some results, otherwise we are at the end of database or something went wrong
297  if (error_code != SQLITE_ROW)
298  {
299  ARMARX_ERROR_S << "Query did not succeed or we got no result while trying to seek: " << sqlite3_errmsg(db);
300  sqlite3_finalize(stmt);
301  sqlite3_close(db);
302  database_open = false;
303  return false;
304  }
305 
306  return true;
307  }
308 
309 } // namespace armarx
armarx::DatabaseTopicReader::DatabaseTopicReader
DatabaseTopicReader(std::filesystem::path path)
Definition: DatabaseTopicReader.cpp:30
DatabaseTopicReader.h
armarx::DatabaseTopicReader::getReplayTopics
std::vector< std::string > getReplayTopics() override
Definition: DatabaseTopicReader.cpp:213
armarx::TopicUtil::TopicData
Definition: TopicUtil.h:34
ARMARX_ERROR_S
#define ARMARX_ERROR_S
Definition: Logging.h:209
data
uint8_t data[1]
Definition: EtherCATFrame.h:68
armarx::DatabaseTopicReader::~DatabaseTopicReader
~DatabaseTopicReader()
Definition: DatabaseTopicReader.cpp:78
armarx::DatabaseTopicReader::read
bool read(TopicUtil::TopicData &data) override
Definition: DatabaseTopicReader.cpp:90
ARMARX_ERROR
#define ARMARX_ERROR
Definition: Logging.h:189
armarx::armem::Time
armarx::core::time::DateTime Time
Definition: forward_declarations.h:13
armarx::DatabaseTopicReader::seekTo
bool seekTo(IceUtil::Time timestamp) override
Definition: DatabaseTopicReader.cpp:145
armarx::armem::server::ltm::mongodb::constantes::ID
const std::string ID
Definition: operations.h:15
ARMARX_INFO
#define ARMARX_INFO
Definition: Logging.h:174
armarx::DatabaseTopicReader::getReplayLength
IceUtil::Time getReplayLength() override
Definition: DatabaseTopicReader.cpp:186
armarx::DatabaseTopicReader::getFilepath
std::filesystem::path getFilepath() const
Definition: DatabaseTopicReader.cpp:85
Logging.h
armarx
This file offers overloads of toIce() and fromIce() functions for STL container types.
Definition: ArmarXTimeserver.cpp:28