1 #include "databasesink.h"
2 #include "abstractroutingengine.h"
3 #include "listplusplus.h"
5 int bufferLength = 100;
8 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
10 return new DatabaseSinkManager(routingengine, config);
13 void * cbFunc(gpointer data)
15 Shared *shared = static_cast<Shared*>(data);
19 throw std::runtime_error("Could not cast shared object.");
22 vector<DictionaryList<string> > insertList;
28 DBObject obj = shared->queue.pop();
35 DictionaryList<string> dict;
37 NameValuePair<string> one("key", obj.key);
38 NameValuePair<string> two("value", obj.value);
39 NameValuePair<string> three("source", obj.source);
40 NameValuePair<string> zone("zone", boost::lexical_cast<string>(obj.zone));
41 NameValuePair<string> four("time", boost::lexical_cast<string>(obj.time));
42 NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj.sequence));
46 dict.push_back(three);
51 insertList.push_back(dict);
53 if(insertList.size() > bufferLength)
55 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
56 for(int i=0; i< insertList.size(); i++)
58 DictionaryList<string> d = insertList[i];
59 shared->db->insert(d);
61 shared->db->exec("END TRANSACTION");
67 /// final flush of whatever is still in the queue:
69 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
70 for(int i=0; i< insertList.size(); i++)
72 DictionaryList<string> d = insertList[i];
73 shared->db->insert(d);
75 shared->db->exec("END TRANSACTION");
80 int getNextEvent(gpointer data)
82 PlaybackShared* pbshared = static_cast<PlaybackShared*>(data);
85 throw std::runtime_error("failed to cast PlaybackShared object");
90 auto itr = pbshared->playbackQueue.begin();
92 if(itr == pbshared->playbackQueue.end())
99 AbstractPropertyType* value = VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key,obj.value);
103 pbshared->routingEngine->updateProperty(value, pbshared->uuid);
104 value->timestamp = obj.time;
105 value->sequence = obj.sequence;
106 value->sourceUuid = obj.source;
107 value->zone = obj.zone;
110 if(++itr != pbshared->playbackQueue.end())
113 double t = o2.time - obj.time;
116 g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared);
118 g_timeout_add(1, getNextEvent, pbshared);
121 pbshared->playbackQueue.remove(obj);
122 DebugOut()<<"playback Queue size: "<<pbshared->playbackQueue.size()<<endl;
128 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config)
129 :AbstractSource(engine,config),thread(NULL),shared(NULL),playback(false),playbackShared(NULL), playbackMultiplier(1)
131 databaseName = "storage";
133 tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, zone REAL, time REAL, sequence REAL)";
135 if(config.find("databaseFile") != config.end())
137 setDatabaseFileName(config["databaseFile"]);
140 if(config.find("bufferLength") != config.end())
142 bufferLength = atoi(config["bufferLength"].c_str());
145 if(config.find("frequency") != config.end())
149 int t = boost::lexical_cast<int>(config["frequency"]);
153 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
159 if(config.find("properties") != config.end())
164 for(auto itr=propertiesToSubscribeTo.begin();itr!=propertiesToSubscribeTo.end();itr++)
166 engine->subscribeToProperty(*itr,this);
169 mSupported.push_back(DatabaseFile);
170 mSupported.push_back(DatabaseLogging);
171 mSupported.push_back(DatabasePlayback);
173 routingEngine->updateSupported(supported(), PropertyList(), this);
175 if(config.find("startOnLoad")!= config.end())
177 setLogging(config["startOnLoad"] == "true");
180 if(config.find("playbackMultiplier")!= config.end())
182 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
185 if(config.find("playbackOnLoad")!= config.end())
187 setPlayback(config["playbackOnLoad"] == "true");
193 DatabaseSink::~DatabaseSink()
202 delete playbackShared;
207 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
212 PropertyList DatabaseSink::supported()
217 PropertyInfo DatabaseSink::getPropertyInfo(VehicleProperty::Property property)
219 /// TODO: Compute update frequency for properties in the database
220 return PropertyInfo::invalid();
223 void DatabaseSink::parseConfig()
225 json_object *rootobject;
226 json_tokener *tokener = json_tokener_new();
227 enum json_tokener_error err;
230 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
231 } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
232 if (err != json_tokener_success)
234 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
236 if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
238 //Should handle the extra data here sometime...
241 json_object *propobject = json_object_object_get(rootobject,"properties");
243 g_assert(json_object_get_type(propobject) == json_type_array);
245 array_list *proplist = json_object_get_array(propobject);
247 for(int i=0; i < array_list_length(proplist); i++)
249 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
250 std::string prop = json_object_get_string(idxobj);
251 propertiesToSubscribeTo.push_back(prop);
253 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
256 //json_object_put(propobject);
257 json_object_put(rootobject);
260 void DatabaseSink::stopDb()
267 shared->queue.append(obj);
269 g_thread_join(thread);
275 void DatabaseSink::startDb()
279 DebugOut(0)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<<endl;
285 DebugOut(0)<<"WARNING: logging already started. doing nothing."<<endl;
291 thread = g_thread_new("dbthread", cbFunc, shared);
294 void DatabaseSink::startPlayback()
303 /// populate playback queue:
305 vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
307 /// we are done with shared. clean up:
313 delete playbackShared;
316 playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
318 for(int i=0;i<results.size();i++)
320 if(results[i].size() < 5)
322 throw std::runtime_error("column mismatch in query");
327 obj.key = results[i][0];
328 obj.value = results[i][1];
329 obj.source = results[i][2];
330 obj.time = boost::lexical_cast<double>(results[i][3]);
332 /// TODO: figure out why sequence is broken:
334 // obj->sequence = boost::lexical_cast<int>(results[i][4]);
336 playbackShared->playbackQueue.push_back(obj);
339 g_timeout_add(0,getNextEvent,playbackShared);
342 void DatabaseSink::initDb()
344 if(shared) delete shared;
347 shared->db->init(databaseName, tablename, tablecreate);
350 void DatabaseSink::setPlayback(bool v)
352 AsyncSetPropertyRequest request;
353 request.property = DatabasePlayback;
354 request.value = new DatabasePlaybackType(v);
356 setProperty(request);
359 void DatabaseSink::setLogging(bool b)
361 AsyncSetPropertyRequest request;
362 request.property = DatabaseLogging;
363 request.value = new DatabaseLoggingType(b);
365 setProperty(request);
367 delete request.value;
370 void DatabaseSink::setDatabaseFileName(string filename)
372 databaseName = filename;
376 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key FROM "+tablename);
378 for(int i=0; i < supportedStr.size(); i++)
380 if(!contains(mSupported, supportedStr[i][0]))
381 mSupported.push_back(supportedStr[i][0]);
387 routingEngine->updateSupported(mSupported, PropertyList(), this);
390 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
392 VehicleProperty::Property property = value->name;
397 if(!contains(mSupported, property))
399 mSupported.push_back(property);
400 routingEngine->updateSupported(mSupported, PropertyList(), this);
405 obj.value = value->toString();
406 obj.source = value->sourceUuid;
407 obj.time = value->timestamp;
408 obj.sequence = value->sequence;
409 obj.zone = value->zone;
411 shared->queue.append(obj);
415 const string DatabaseSink::uuid()
417 return "9f88156e-cb92-4472-8775-9c08addf50d3";
420 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
422 reply->success = false;
424 if(reply->property == DatabaseFile)
426 DatabaseFileType temp(databaseName);
427 reply->value = &temp;
429 reply->success = true;
430 reply->completed(reply);
434 else if(reply->property == DatabaseLogging)
436 DatabaseLoggingType temp = shared;
438 reply->value = &temp;
439 reply->success = true;
440 reply->completed(reply);
445 else if(reply->property == DatabasePlayback)
447 DatabasePlaybackType temp = playback;
448 reply->value = &temp;
449 reply->success = true;
450 reply->completed(reply);
455 reply->completed(reply);
458 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
460 BaseDB * db = new BaseDB();
461 db->init(databaseName, tablename, tablecreate);
466 query<<"SELECT * from "<<tablename<<" WHERE (";
468 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
470 if(itr != reply->properties.begin())
473 query<<"key='"<<(*itr)<<"'";
478 if(reply->timeBegin && reply->timeEnd)
480 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
483 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
485 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
488 if(reply->sourceUuid != "")
489 query<<" AND source='"<<reply->sourceUuid<<"'";
491 query<<" AND zone="<<reply->zone;
493 std::vector<std::vector<string>> data = db->select(query.str());
495 DebugOut()<<"Dataset size "<<data.size()<<endl;
497 for(auto i=0;i<data.size();i++)
499 if(data[i].size() != 6)
503 dbobj.key = data[i][0];
504 dbobj.value = data[i][1];
505 dbobj.source = data[i][2];
506 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
507 dbobj.time = boost::lexical_cast<double>(data[i][4]);
508 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
510 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
513 property->timestamp = dbobj.time;
514 property->sequence = dbobj.sequence;
516 reply->values.push_back(property);
520 reply->success = true;
521 reply->completed(reply);
526 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
528 AsyncPropertyReply* reply = new AsyncPropertyReply(request);
529 reply->success = false;
531 if(request.property == DatabaseLogging)
533 if(request.value->value<bool>())
537 reply->success = true;
538 DatabaseLoggingType temp(true);
539 routingEngine->updateProperty(&temp,uuid());
544 reply->success = true;
545 DatabaseLoggingType temp(false);
546 routingEngine->updateProperty(&temp,uuid());
550 else if(request.property == DatabaseFile)
552 std::string fname = request.value->toString();
554 databaseName = fname;
556 DatabaseFileType temp(databaseName);
558 routingEngine->updateProperty(&temp,uuid());
560 reply->success = true;
562 else if( request.property == DatabasePlayback)
564 if(request.value->value<bool>())
569 DatabasePlaybackType temp(playback);
571 routingEngine->updateProperty(&temp,uuid());
576 playbackShared->stop = true;
580 DatabasePlaybackType temp(playback);
582 routingEngine->updateProperty(&temp,uuid());
585 reply->success = true;
591 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
596 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )