1 #include "databasesink.h"
2 #include "abstractroutingengine.h"
3 #include "listplusplus.h"
4 #include "superptr.hpp"
5 #include "uuidhelper.h"
10 int bufferLength = 100;
13 extern "C" AbstractSource * create(AbstractRoutingEngine* routingengine, map<string, string> config)
15 auto plugin = new AmbPlugin<DatabaseSink>(routingengine, config);
21 void * cbFunc(Shared* shared)
25 throw std::runtime_error("Could not cast shared object.");
29 shared->tripId = amb::createUuid();
31 vector<DictionaryList<string> > insertList;
37 DBObject obj = shared->queue.pop();
44 DictionaryList<string> dict;
46 NameValuePair<string> one("key", obj.key);
47 NameValuePair<string> two("value", obj.value);
48 NameValuePair<string> three("source", obj.source);
49 NameValuePair<string> zone("zone", boost::lexical_cast<string>(obj.zone));
50 NameValuePair<string> four("time", boost::lexical_cast<string>(obj.time));
51 NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj.sequence));
52 NameValuePair<string> six("tripId", boost::lexical_cast<string>(shared->tripId));
56 dict.push_back(three);
62 insertList.push_back(dict);
64 if(insertList.size() > bufferLength)
66 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
67 for(int i=0; i< insertList.size(); i++)
69 DictionaryList<string> d = insertList[i];
70 shared->db->insert(d);
72 shared->db->exec("END TRANSACTION");
78 /// final flush of whatever is still in the queue:
80 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
81 for(int i=0; i< insertList.size(); i++)
83 DictionaryList<string> d = insertList[i];
84 shared->db->insert(d);
86 shared->db->exec("END TRANSACTION");
91 int getNextEvent(gpointer data)
93 PlaybackShared* pbshared = static_cast<PlaybackShared*>(data);
96 throw std::runtime_error("failed to cast PlaybackShared object");
101 auto itr = pbshared->playbackQueue.begin();
103 if(itr == pbshared->playbackQueue.end())
110 auto value = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key, obj.value));
114 value->priority = AbstractPropertyType::Instant;
115 value->timestamp = obj.time;
116 value->sequence = obj.sequence;
117 value->sourceUuid = obj.source;
118 value->zone = obj.zone;
119 pbshared->routingEngine->updateProperty(value.get(), pbshared->uuid);
122 if(++itr != pbshared->playbackQueue.end())
125 double t = o2.time - obj.time;
128 g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared);
130 g_timeout_add(1, getNextEvent, pbshared);
133 pbshared->playbackQueue.remove(obj);
134 DebugOut()<<"playback Queue size: "<<pbshared->playbackQueue.size()<<endl;
139 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config, AbstractSource &parent)
140 :AmbPluginImpl(engine, config, parent), shared(nullptr), playback(false), playbackShared(nullptr), playbackMultiplier(1)
142 databaseName = "storage";
144 tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, zone REAL, time REAL, sequence REAL, tripId TEXT)";
146 if(config.find("databaseFile") != config.end())
148 setDatabaseFileName(config["databaseFile"]);
151 if(config.find("bufferLength") != config.end())
153 bufferLength = atoi(config["bufferLength"].c_str());
156 if(config.find("frequency") != config.end())
160 int t = boost::lexical_cast<int>(config["frequency"]);
164 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
170 if(config.find("properties") != config.end())
175 for(auto itr : propertiesToSubscribeTo)
177 engine->subscribeToProperty(itr, &parent);
180 addPropertySupport(Zone::None, [](){return new DatabaseFileType("storage");});
181 addPropertySupport(Zone::None, [](){return new DatabasePlaybackType(false);});
182 addPropertySupport(Zone::None, [](){return new DatabaseLoggingType(false);});
184 routingEngine->updateSupported(supported(), PropertyList(), &parent);
186 if(config.find("startOnLoad")!= config.end())
188 setLogging(config["startOnLoad"] == "true");
191 if(config.find("playbackMultiplier")!= config.end())
193 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
196 if(config.find("playbackOnLoad")!= config.end())
198 setPlayback(config["playbackOnLoad"] == "true");
204 DatabaseSink::~DatabaseSink()
213 delete playbackShared;
218 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
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);
275 void DatabaseSink::startDb()
277 if(playback.basicValue())
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 = std::thread(cbFunc, shared);
294 void DatabaseSink::startPlayback()
296 if(playback.basicValue())
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.zone = boost::lexical_cast<double>(results[i][3]);
331 obj.time = boost::lexical_cast<double>(results[i][4]);
332 obj.sequence = boost::lexical_cast<double>(results[i][5]);
334 playbackShared->playbackQueue.push_back(obj);
337 g_timeout_add(0, getNextEvent, playbackShared);
340 void DatabaseSink::initDb()
342 if(shared) delete shared;
345 shared->db->init(databaseName.value<std::string>(), tablename, tablecreate);
348 void DatabaseSink::setPlayback(bool v)
350 AsyncSetPropertyRequest request;
351 request.property = DatabasePlayback;
352 request.value = new DatabasePlaybackType(v);
354 setProperty(request);
357 void DatabaseSink::setLogging(bool b)
360 AsyncSetPropertyRequest request;
361 request.property = DatabaseLogging;
362 request.value = &databaseLogging;
364 setProperty(request);
367 void DatabaseSink::setDatabaseFileName(string filename)
369 databaseName = filename;
373 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key FROM "+tablename);
375 for(int i=0; i < supportedStr.size(); i++)
377 if(!contains(supported(), supportedStr[i][0]))
379 std::string name = supportedStr[i][0];
380 Zone::Type zone = boost::lexical_cast<int, std::string>(supportedStr[i][3]);
381 addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); });
388 routingEngine->updateSupported(mSupported, PropertyList(), &source);
391 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
393 VehicleProperty::Property property = value->name;
398 if(!contains(supported(), property))
400 addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);});
401 routingEngine->updateSupported(mSupported, PropertyList(), &source);
406 obj.value = value->toString();
407 obj.source = value->sourceUuid;
408 obj.time = value->timestamp;
409 obj.sequence = value->sequence;
410 obj.zone = value->zone;
412 shared->queue.append(obj);
416 const std::string DatabaseSink::uuid() const
418 return "9f88156e-cb92-4472-8775-9c08addf50d3";
421 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
423 reply->success = false;
425 if(reply->property == DatabaseFile)
427 DatabaseFileType temp(databaseName);
428 reply->value = &temp;
430 reply->success = true;
431 reply->completed(reply);
435 else if(reply->property == DatabaseLogging)
437 databaseLogging = shared != nullptr;
439 reply->value = &databaseLogging;
440 reply->success = true;
441 reply->completed(reply);
446 else if(reply->property == DatabasePlayback)
448 DatabasePlaybackType temp = playback;
449 reply->value = &temp;
450 reply->success = true;
451 reply->completed(reply);
456 reply->completed(reply);
459 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
461 BaseDB * db = new BaseDB();
462 db->init(databaseName.value<std::string>(), tablename, tablecreate);
467 query<<"SELECT * from "<<tablename<<" WHERE (";
469 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
471 if(itr != reply->properties.begin())
474 query<<"key='"<<(*itr)<<"'";
479 if(reply->timeBegin && reply->timeEnd)
481 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
484 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
486 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
489 if(reply->sourceUuid != "")
490 query<<" AND source='"<<reply->sourceUuid<<"'";
492 query<<" AND zone="<<reply->zone;
494 std::vector<std::vector<string>> data = db->select(query.str());
496 DebugOut()<<"Dataset size "<<data.size()<<endl;
498 for(auto i=0;i<data.size();i++)
500 if(data[i].size() != 6)
504 dbobj.key = data[i][0];
505 dbobj.value = data[i][1];
506 dbobj.source = data[i][2];
507 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
508 dbobj.time = boost::lexical_cast<double>(data[i][4]);
509 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
511 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
514 property->timestamp = dbobj.time;
515 property->sequence = dbobj.sequence;
517 reply->values.push_back(property);
521 reply->success = true;
522 reply->completed(reply);
527 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
529 AsyncPropertyReply* reply = new AsyncPropertyReply(request);
530 reply->success = false;
532 if(request.property == DatabaseLogging)
534 if(request.value->value<bool>())
538 reply->success = true;
539 databaseLogging = true;
540 routingEngine->updateProperty(&databaseLogging,uuid());
545 reply->success = true;
546 databaseLogging = false;
547 routingEngine->updateProperty(&databaseLogging,uuid());
551 else if(request.property == DatabaseFile)
553 std::string fname = request.value->toString();
555 databaseName = fname;
557 routingEngine->updateProperty(&databaseName,uuid());
559 reply->success = true;
561 else if( request.property == DatabasePlayback)
563 if(request.value->value<bool>())
568 routingEngine->updateProperty(&playback,uuid());
573 playbackShared->stop = true;
577 routingEngine->updateProperty(&playback,uuid());
580 reply->success = true;
586 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
591 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )