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 static 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)
143 tablecreate = database_table_create;
145 if(config.find("bufferLength") != config.end())
147 bufferLength = boost::lexical_cast<int>(config["bufferLength"]);
150 if(config.find("frequency") != config.end())
154 int t = boost::lexical_cast<int>(config["frequency"]);
158 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
162 if(config.find("properties") != config.end())
167 for(auto itr : propertiesToSubscribeTo)
169 engine->subscribeToProperty(itr, &parent);
172 databaseName = addPropertySupport(Zone::None, [](){ return new DatabaseFileType("storage"); });
173 playback = addPropertySupport(Zone::None, [](){ return new DatabasePlaybackType(false); });
174 databaseLogging = addPropertySupport(Zone::None, [](){ return new DatabaseLoggingType(false); });
176 if(config.find("startOnLoad")!= config.end())
178 databaseLogging->setValue(config["startOnLoad"] == "true");
181 if(config.find("playbackMultiplier")!= config.end())
183 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
186 if(config.find("playbackOnLoad")!= config.end())
188 playback->setValue(config["playbackOnLoad"] == "true");
192 DatabaseSink::~DatabaseSink()
201 delete playbackShared;
206 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
211 void DatabaseSink::parseConfig()
213 json_object *rootobject;
214 json_tokener *tokener = json_tokener_new();
215 enum json_tokener_error err;
218 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
219 } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
220 if (err != json_tokener_success)
222 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
224 if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
226 //Should handle the extra data here sometime...
229 json_object *propobject = json_object_object_get(rootobject,"properties");
231 g_assert(json_object_get_type(propobject) == json_type_array);
233 array_list *proplist = json_object_get_array(propobject);
235 for(int i=0; i < array_list_length(proplist); i++)
237 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
238 std::string prop = json_object_get_string(idxobj);
239 propertiesToSubscribeTo.push_back(prop);
241 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
244 //json_object_put(propobject);
245 json_object_put(rootobject);
248 void DatabaseSink::stopDb()
255 shared->queue.append(obj);
257 if(thread && thread->joinable())
264 void DatabaseSink::startDb()
266 if(playback->value<bool>())
268 DebugOut(0)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<<endl;
274 DebugOut(0)<<"WARNING: logging already started. doing nothing."<<endl;
280 if(thread && thread->joinable())
283 thread = amb::make_unique(new std::thread(cbFunc, shared));
286 void DatabaseSink::startPlayback()
288 if(playback->value<bool>())
291 playback->setValue(true);
295 /// populate playback queue:
297 vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
299 /// we are done with shared. clean up:
305 delete playbackShared;
308 playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
310 for(int i=0;i<results.size();i++)
312 if(results[i].size() < 5)
314 throw std::runtime_error("column mismatch in query");
319 obj.key = results[i][0];
320 obj.value = results[i][1];
321 obj.source = results[i][2];
322 obj.zone = boost::lexical_cast<int>(results[i][3]);
323 obj.time = boost::lexical_cast<double>(results[i][4]);
324 obj.sequence = boost::lexical_cast<double>(results[i][5]);
326 playbackShared->playbackQueue.push_back(obj);
329 g_timeout_add(0, getNextEvent, playbackShared);
332 void DatabaseSink::initDb()
334 if(shared) delete shared;
337 shared->db->init(databaseName->value<std::string>(), tablename, tablecreate);
340 void DatabaseSink::setDatabaseFileName(string filename)
344 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key, zone, source FROM " + tablename);
346 for(int i=0; i < supportedStr.size(); i++)
348 std::string name = supportedStr[i][0];
350 if(!contains(supported(), name))
352 std::string zoneStr = supportedStr[i][1];
353 std::string sourceStr = supportedStr[i][2];
355 DebugOut() << "adding property " << name << " in zone: " << zoneStr << "for source: " << sourceStr << endl;
357 Zone::Type zone = boost::lexical_cast<Zone::Type>(zoneStr);
358 auto property = addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); });
359 property->sourceUuid = sourceStr;
366 routingEngine->updateSupported(supported(), PropertyList(), &source);
369 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
371 VehicleProperty::Property property = value->name;
376 if(!contains(supported(), property))
378 addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);});
379 routingEngine->updateSupported(supported(), PropertyList(), &source);
384 obj.value = value->toString();
385 obj.source = value->sourceUuid;
386 obj.time = value->timestamp;
387 obj.sequence = value->sequence;
388 obj.zone = value->zone;
390 shared->queue.append(obj);
394 const std::string DatabaseSink::uuid() const
396 return "9f88156e-cb92-4472-8775-9c08addf50d3";
399 void DatabaseSink::init()
401 if(configuration.find("databaseFile") != configuration.end())
403 databaseName->setValue(configuration["databaseFile"]);
404 setDatabaseFileName(configuration["databaseFile"]);
407 routingEngine->updateSupported(supported(), PropertyList(), &source);
410 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
412 BaseDB * db = new BaseDB();
413 db->init(databaseName->value<std::string>(), tablename, tablecreate);
418 query<<"SELECT * from "<<tablename<<" WHERE (";
420 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
422 DebugOut() << "prop: " << (*itr) << endl;
423 if(itr != reply->properties.begin())
426 query<<"key='"<<(*itr)<<"'";
433 query<<" AND time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
436 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
438 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
441 if(reply->sourceUuid != "")
442 query<<" AND source='"<<reply->sourceUuid<<"'";
444 query<<" AND zone="<<reply->zone;
446 std::vector<std::vector<string>> data = db->select(query.str());
448 DebugOut()<<"Dataset size "<<data.size()<<endl;
452 reply->success = false;
453 reply->error = AsyncPropertyReply::InvalidOperation;
457 reply->success = true;
460 for(auto i=0; i<data.size(); i++)
462 if(data[i].size() != 7)
466 dbobj.key = data[i][0];
467 dbobj.value = data[i][1];
468 dbobj.source = data[i][2];
469 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
470 dbobj.time = boost::lexical_cast<double>(data[i][4]);
471 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
472 dbobj.tripId = data[i][6];
474 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
477 property->timestamp = dbobj.time;
478 property->sequence = dbobj.sequence;
480 reply->values.push_back(property);
485 reply->completed(reply);
490 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
492 AsyncPropertyReply* reply = AmbPluginImpl::setProperty(request);
494 if(request.property == DatabaseLogging)
496 if(request.value->value<bool>())
505 else if(request.property == DatabaseFile)
507 setDatabaseFileName(databaseName->value<std::string>());
509 else if( request.property == DatabasePlayback)
511 if(request.value->value<bool>())
518 playbackShared->stop = true;
527 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
532 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )