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 INTEGER, time REAL, sequence REAL, tripId TEXT)";
146 if(config.find("bufferLength") != config.end())
148 bufferLength = atoi(config["bufferLength"].c_str());
151 if(config.find("frequency") != config.end())
155 int t = boost::lexical_cast<int>(config["frequency"]);
159 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
165 if(config.find("properties") != config.end())
170 for(auto itr : propertiesToSubscribeTo)
172 engine->subscribeToProperty(itr, &parent);
175 addPropertySupport(Zone::None, [](){ return new DatabaseFileType("storage"); });
176 addPropertySupport(Zone::None, [](){ return new DatabasePlaybackType(false); });
177 addPropertySupport(Zone::None, [](){ return new DatabaseLoggingType(false); });
179 if(config.find("startOnLoad")!= config.end())
181 setLogging(config["startOnLoad"] == "true");
184 if(config.find("playbackMultiplier")!= config.end())
186 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
189 if(config.find("playbackOnLoad")!= config.end())
191 setPlayback(config["playbackOnLoad"] == "true");
195 DatabaseSink::~DatabaseSink()
204 delete playbackShared;
209 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
214 void DatabaseSink::parseConfig()
216 json_object *rootobject;
217 json_tokener *tokener = json_tokener_new();
218 enum json_tokener_error err;
221 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
222 } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
223 if (err != json_tokener_success)
225 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
227 if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
229 //Should handle the extra data here sometime...
232 json_object *propobject = json_object_object_get(rootobject,"properties");
234 g_assert(json_object_get_type(propobject) == json_type_array);
236 array_list *proplist = json_object_get_array(propobject);
238 for(int i=0; i < array_list_length(proplist); i++)
240 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
241 std::string prop = json_object_get_string(idxobj);
242 propertiesToSubscribeTo.push_back(prop);
244 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
247 //json_object_put(propobject);
248 json_object_put(rootobject);
251 void DatabaseSink::stopDb()
258 shared->queue.append(obj);
266 void DatabaseSink::startDb()
268 if(playback.basicValue())
270 DebugOut(0)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<<endl;
276 DebugOut(0)<<"WARNING: logging already started. doing nothing."<<endl;
282 thread = std::thread(cbFunc, shared);
285 void DatabaseSink::startPlayback()
287 if(playback.basicValue())
294 /// populate playback queue:
296 vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
298 /// we are done with shared. clean up:
304 delete playbackShared;
307 playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
309 for(int i=0;i<results.size();i++)
311 if(results[i].size() < 5)
313 throw std::runtime_error("column mismatch in query");
318 obj.key = results[i][0];
319 obj.value = results[i][1];
320 obj.source = results[i][2];
321 obj.zone = boost::lexical_cast<int>(results[i][3]);
322 obj.time = boost::lexical_cast<double>(results[i][4]);
323 obj.sequence = boost::lexical_cast<double>(results[i][5]);
325 playbackShared->playbackQueue.push_back(obj);
328 g_timeout_add(0, getNextEvent, playbackShared);
331 void DatabaseSink::initDb()
333 if(shared) delete shared;
336 shared->db->init(databaseName.value<std::string>(), tablename, tablecreate);
339 void DatabaseSink::setPlayback(bool v)
341 AsyncSetPropertyRequest request;
342 request.property = DatabasePlayback;
343 request.value = new DatabasePlaybackType(v);
345 setProperty(request);
348 void DatabaseSink::setLogging(bool b)
351 AsyncSetPropertyRequest request;
352 request.property = DatabaseLogging;
353 request.value = &databaseLogging;
355 setProperty(request);
358 void DatabaseSink::setDatabaseFileName(string filename)
360 databaseName = filename;
364 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key, zone FROM "+tablename);
366 for(int i=0; i < supportedStr.size(); i++)
368 std::string name = supportedStr[i][0];
370 if(!contains(supported(), name))
372 std::string zoneStr = supportedStr[i][1];
374 DebugOut() << "adding property " << name << " in zone: " << zoneStr << endl;
376 Zone::Type zone = boost::lexical_cast<Zone::Type>(zoneStr);
377 addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); });
384 routingEngine->updateSupported(supported(), PropertyList(), &source);
387 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
389 VehicleProperty::Property property = value->name;
394 if(!contains(supported(), property))
396 addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);});
397 routingEngine->updateSupported(supported(), PropertyList(), &source);
402 obj.value = value->toString();
403 obj.source = value->sourceUuid;
404 obj.time = value->timestamp;
405 obj.sequence = value->sequence;
406 obj.zone = value->zone;
408 shared->queue.append(obj);
412 const std::string DatabaseSink::uuid() const
414 return "9f88156e-cb92-4472-8775-9c08addf50d3";
417 void DatabaseSink::init()
419 if(configuration.find("databaseFile") != configuration.end())
421 setDatabaseFileName(configuration["databaseFile"]);
424 routingEngine->updateSupported(supported(), PropertyList(), &source);
427 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
429 reply->success = false;
431 if(reply->property == DatabaseFile)
433 DatabaseFileType temp(databaseName);
434 reply->value = &temp;
436 reply->success = true;
437 reply->completed(reply);
441 else if(reply->property == DatabaseLogging)
443 databaseLogging = shared != nullptr;
445 reply->value = &databaseLogging;
446 reply->success = true;
447 reply->completed(reply);
452 else if(reply->property == DatabasePlayback)
454 DatabasePlaybackType temp = playback;
455 reply->value = &temp;
456 reply->success = true;
457 reply->completed(reply);
462 reply->completed(reply);
465 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
467 BaseDB * db = new BaseDB();
468 db->init(databaseName.value<std::string>(), tablename, tablecreate);
473 query<<"SELECT * from "<<tablename<<" WHERE (";
475 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
477 if(itr != reply->properties.begin())
480 query<<"key='"<<(*itr)<<"'";
485 if(reply->timeBegin && reply->timeEnd)
487 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
490 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
492 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
495 if(reply->sourceUuid != "")
496 query<<" AND source='"<<reply->sourceUuid<<"'";
498 query<<" AND zone="<<reply->zone;
500 std::vector<std::vector<string>> data = db->select(query.str());
502 DebugOut()<<"Dataset size "<<data.size()<<endl;
504 for(auto i=0;i<data.size();i++)
506 if(data[i].size() != 6)
510 dbobj.key = data[i][0];
511 dbobj.value = data[i][1];
512 dbobj.source = data[i][2];
513 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
514 dbobj.time = boost::lexical_cast<double>(data[i][4]);
515 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
517 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
520 property->timestamp = dbobj.time;
521 property->sequence = dbobj.sequence;
523 reply->values.push_back(property);
527 reply->success = true;
528 reply->completed(reply);
533 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
535 AsyncPropertyReply* reply = new AsyncPropertyReply(request);
536 reply->success = false;
538 if(request.property == DatabaseLogging)
540 if(request.value->value<bool>())
544 reply->success = true;
545 databaseLogging = true;
546 routingEngine->updateProperty(&databaseLogging,uuid());
551 reply->success = true;
552 databaseLogging = false;
553 routingEngine->updateProperty(&databaseLogging,uuid());
557 else if(request.property == DatabaseFile)
559 std::string fname = request.value->toString();
561 databaseName = fname;
563 routingEngine->updateProperty(&databaseName,uuid());
565 reply->success = true;
567 else if( request.property == DatabasePlayback)
569 if(request.value->value<bool>())
574 routingEngine->updateProperty(&playback,uuid());
579 playbackShared->stop = true;
583 routingEngine->updateProperty(&playback,uuid());
586 reply->success = true;
592 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
597 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )