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" void create(AbstractRoutingEngine* routingengine, map<string, string> config)
15 auto plugin = new AmbPlugin<DatabaseSink>(routingengine, config);
19 static void * cbFunc(Shared* shared)
23 throw std::runtime_error("Could not cast shared object.");
27 shared->tripId = amb::createUuid();
29 vector<DictionaryList<string> > insertList;
35 DBObject obj = shared->queue.pop();
42 DictionaryList<string> dict;
44 NameValuePair<string> one("key", obj.key);
45 NameValuePair<string> two("value", obj.value);
46 NameValuePair<string> three("source", obj.source);
47 NameValuePair<string> zone("zone", boost::lexical_cast<string>(obj.zone));
48 NameValuePair<string> four("time", boost::lexical_cast<string>(obj.time));
49 NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj.sequence));
50 NameValuePair<string> six("tripId", boost::lexical_cast<string>(shared->tripId));
54 dict.push_back(three);
60 insertList.push_back(dict);
62 if(insertList.size() > bufferLength)
64 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
65 for(int i=0; i< insertList.size(); i++)
67 DictionaryList<string> d = insertList[i];
68 shared->db->insert(d);
70 shared->db->exec("END TRANSACTION");
76 /// final flush of whatever is still in the queue:
78 shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
79 for(int i=0; i< insertList.size(); i++)
81 DictionaryList<string> d = insertList[i];
82 shared->db->insert(d);
84 shared->db->exec("END TRANSACTION");
89 int getNextEvent(gpointer data)
91 PlaybackShared* pbshared = static_cast<PlaybackShared*>(data);
94 throw std::runtime_error("failed to cast PlaybackShared object");
99 auto itr = pbshared->playbackQueue.begin();
101 if(itr == pbshared->playbackQueue.end())
108 auto value = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key, obj.value));
112 value->priority = AbstractPropertyType::Instant;
113 value->timestamp = obj.time;
114 value->sequence = obj.sequence;
115 value->sourceUuid = obj.source;
116 value->zone = obj.zone;
117 pbshared->routingEngine->updateProperty(value.get(), pbshared->uuid);
120 if(++itr != pbshared->playbackQueue.end())
123 double t = o2.time - obj.time;
126 g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared);
128 g_timeout_add(1, getNextEvent, pbshared);
131 pbshared->playbackQueue.remove(obj);
132 DebugOut()<<"playback Queue size: "<<pbshared->playbackQueue.size()<<endl;
137 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config, AbstractSource &parent)
138 :AmbPluginImpl(engine, config, parent), shared(nullptr), playback(false), playbackShared(nullptr), playbackMultiplier(1)
141 tablecreate = database_table_create;
143 if(config.find("bufferLength") != config.end())
145 bufferLength = boost::lexical_cast<int>(config["bufferLength"]);
148 if(config.find("frequency") != config.end())
152 int t = boost::lexical_cast<int>(config["frequency"]);
156 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
160 if(config.find("properties") != config.end())
165 for(auto itr : propertiesToSubscribeTo)
167 engine->subscribeToProperty(itr, &parent);
170 databaseName = addPropertySupport(Zone::None, [](){ return new DatabaseFileType("storage"); });
171 playback = addPropertySupport(Zone::None, [](){ return new DatabasePlaybackType(false); });
172 databaseLogging = addPropertySupport(Zone::None, [](){ return new DatabaseLoggingType(false); });
174 if(config.find("startOnLoad")!= config.end())
176 databaseLogging->setValue(config["startOnLoad"] == "true");
179 if(config.find("playbackMultiplier")!= config.end())
181 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
184 if(config.find("playbackOnLoad")!= config.end())
186 playback->setValue(config["playbackOnLoad"] == "true");
190 DatabaseSink::~DatabaseSink()
199 delete playbackShared;
204 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
209 void DatabaseSink::parseConfig()
211 json_object *rootobject;
212 json_tokener *tokener = json_tokener_new();
213 enum json_tokener_error err;
216 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
217 } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
218 if (err != json_tokener_success)
220 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
222 if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
224 //Should handle the extra data here sometime...
227 json_object *propobject = json_object_object_get(rootobject,"properties");
229 g_assert(json_object_get_type(propobject) == json_type_array);
231 array_list *proplist = json_object_get_array(propobject);
233 for(int i=0; i < array_list_length(proplist); i++)
235 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
236 std::string prop = json_object_get_string(idxobj);
237 propertiesToSubscribeTo.push_back(prop);
239 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
242 //json_object_put(propobject);
243 json_object_put(rootobject);
246 void DatabaseSink::stopDb()
253 shared->queue.append(obj);
255 if(thread && thread->joinable())
262 void DatabaseSink::startDb()
264 if(playback->value<bool>())
266 DebugOut(0)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<<endl;
272 DebugOut(0)<<"WARNING: logging already started. doing nothing."<<endl;
278 if(thread && thread->joinable())
281 thread = amb::make_unique(new std::thread(cbFunc, shared));
284 void DatabaseSink::startPlayback()
286 if(playback->value<bool>())
289 playback->setValue(true);
293 /// populate playback queue:
295 vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
297 /// we are done with shared. clean up:
303 delete playbackShared;
306 playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
308 for(int i=0;i<results.size();i++)
310 if(results[i].size() < 5)
312 throw std::runtime_error("column mismatch in query");
317 obj.key = results[i][0];
318 obj.value = results[i][1];
319 obj.source = results[i][2];
320 obj.zone = boost::lexical_cast<int>(results[i][3]);
321 obj.time = boost::lexical_cast<double>(results[i][4]);
322 obj.sequence = boost::lexical_cast<double>(results[i][5]);
324 playbackShared->playbackQueue.push_back(obj);
327 g_timeout_add(0, getNextEvent, playbackShared);
330 void DatabaseSink::initDb()
332 if(shared) delete shared;
335 shared->db->init(databaseName->value<std::string>(), tablename, tablecreate);
338 void DatabaseSink::setDatabaseFileName(string filename)
342 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key, zone, source FROM " + tablename);
344 for(int i=0; i < supportedStr.size(); i++)
346 std::string name = supportedStr[i][0];
348 if(!contains(supported(), name))
350 std::string zoneStr = supportedStr[i][1];
351 std::string sourceStr = supportedStr[i][2];
353 DebugOut() << "adding property " << name << " in zone: " << zoneStr << "for source: " << sourceStr << endl;
355 Zone::Type zone = boost::lexical_cast<Zone::Type>(zoneStr);
356 auto property = addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); });
357 property->sourceUuid = sourceStr;
364 routingEngine->updateSupported(supported(), PropertyList(), &source);
367 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
369 VehicleProperty::Property property = value->name;
374 if(!contains(supported(), property))
376 addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);});
377 routingEngine->updateSupported(supported(), PropertyList(), &source);
382 obj.value = value->toString();
383 obj.source = value->sourceUuid;
384 obj.time = value->timestamp;
385 obj.sequence = value->sequence;
386 obj.zone = value->zone;
388 shared->queue.append(obj);
392 const std::string DatabaseSink::uuid() const
394 return "9f88156e-cb92-4472-8775-9c08addf50d3";
397 void DatabaseSink::init()
399 if(configuration.find("databaseFile") != configuration.end())
401 databaseName->setValue(configuration["databaseFile"]);
402 setDatabaseFileName(configuration["databaseFile"]);
405 routingEngine->updateSupported(supported(), PropertyList(), &source);
408 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
410 BaseDB * db = new BaseDB();
411 db->init(databaseName->value<std::string>(), tablename, tablecreate);
416 query<<"SELECT * from "<<tablename<<" WHERE (";
418 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
420 DebugOut() << "prop: " << (*itr) << endl;
421 if(itr != reply->properties.begin())
424 query<<"key='"<<(*itr)<<"'";
431 query<<" AND time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
434 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
436 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
439 if(reply->sourceUuid != "")
440 query<<" AND source='"<<reply->sourceUuid<<"'";
442 query<<" AND zone="<<reply->zone;
444 std::vector<std::vector<string>> data = db->select(query.str());
446 DebugOut()<<"Dataset size "<<data.size()<<endl;
450 reply->success = false;
451 reply->error = AsyncPropertyReply::InvalidOperation;
455 reply->success = true;
458 for(auto i=0; i<data.size(); i++)
460 if(data[i].size() != 7)
464 dbobj.key = data[i][0];
465 dbobj.value = data[i][1];
466 dbobj.source = data[i][2];
467 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
468 dbobj.time = boost::lexical_cast<double>(data[i][4]);
469 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
470 dbobj.tripId = data[i][6];
472 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
475 property->timestamp = dbobj.time;
476 property->sequence = dbobj.sequence;
478 reply->values.push_back(property);
483 reply->completed(reply);
488 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
490 AsyncPropertyReply* reply = AmbPluginImpl::setProperty(request);
492 if(request.property == DatabaseLogging)
494 if(request.value->value<bool>())
503 else if(request.property == DatabaseFile)
505 setDatabaseFileName(databaseName->value<std::string>());
507 else if( request.property == DatabasePlayback)
509 if(request.value->value<bool>())
516 playbackShared->stop = true;
525 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
530 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )