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 get shared object.");
27 shared->tripId = amb::createUuid();
29 vector<DictionaryList<string> > insertList;
31 double startTime = amb::currentTime();
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>(amb::Timestamp::instance()->epochTime(obj.time)));
49 NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj.sequence));
50 NameValuePair<string> six("tripId", shared->tripId);
54 dict.push_back(three);
60 insertList.push_back(dict);
62 if(insertList.size() >= bufferLength && amb::currentTime() - startTime >= timeout / 1000)
64 startTime = amb::currentTime();
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(auto d : insertList)
83 shared->db->insert(d);
85 shared->db->exec("END TRANSACTION");
90 int getNextEvent(gpointer data)
92 PlaybackShared* pbshared = static_cast<PlaybackShared*>(data);
95 throw std::runtime_error("failed to cast PlaybackShared object");
100 auto itr = pbshared->playbackQueue.begin();
102 if(itr == pbshared->playbackQueue.end())
109 auto value = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key, obj.value));
113 value->priority = AbstractPropertyType::Instant;
114 value->timestamp = obj.time;
115 value->sequence = obj.sequence;
116 value->sourceUuid = obj.source;
117 value->zone = obj.zone;
118 pbshared->routingEngine->updateProperty(value.get(), pbshared->uuid);
121 if(++itr != pbshared->playbackQueue.end())
124 double t = o2.time - obj.time;
127 g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared);
129 g_timeout_add(1, getNextEvent, pbshared);
132 pbshared->playbackQueue.remove(obj);
133 DebugOut()<<"playback Queue size: "<<pbshared->playbackQueue.size()<<endl;
138 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config, AbstractSource &parent)
139 :AmbPluginImpl(engine, config, parent), shared(nullptr), playback(false), playbackShared(nullptr), playbackMultiplier(1)
142 tablecreate = database_table_create;
144 if(config.find("bufferLength") != config.end())
146 bufferLength = boost::lexical_cast<int>(config["bufferLength"]);
149 if(config.find("frequency") != config.end())
153 int t = boost::lexical_cast<int>(config["frequency"]);
157 DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
161 if(config.find("properties") != config.end())
166 for(auto itr : propertiesToSubscribeTo)
168 engine->subscribeToProperty(itr, &parent);
171 databaseName = addPropertySupport(Zone::None, [](){ return new DatabaseFileType("storage"); });
172 playback = addPropertySupport(Zone::None, [](){ return new DatabasePlaybackType(false); });
173 databaseLogging = addPropertySupport(Zone::None, [](){ return new DatabaseLoggingType(false); });
175 if(config.find("startOnLoad")!= config.end())
177 DebugOut() << "start on load? " << config["startOnLoad"] << endl;
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()
250 databaseLogging->setValue(false);
257 shared->queue.append(obj);
259 if(thread && thread->joinable())
265 routingEngine->updateProperty(databaseLogging.get(), source.uuid());
268 void DatabaseSink::startDb()
270 if(playback->value<bool>())
272 DebugOut(DebugOut::Error)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<<endl;
278 DebugOut(DebugOut::Warning)<<"WARNING: logging already started. doing nothing."<<endl;
284 if(thread && thread->joinable())
287 thread = amb::make_unique(new std::thread(cbFunc, shared));
289 databaseLogging->setValue(true);
290 routingEngine->updateProperty(databaseLogging.get(), source.uuid());
293 void DatabaseSink::startPlayback()
295 if(playback->value<bool>())
298 playback->setValue(true);
302 /// populate playback queue:
304 vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
310 delete playbackShared;
313 playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
315 for(int i=0;i<results.size();i++)
317 if(results[i].size() < 5)
319 throw std::runtime_error("column mismatch in query");
324 obj.key = results[i][0];
325 obj.value = results[i][1];
326 obj.source = results[i][2];
327 obj.zone = boost::lexical_cast<int>(results[i][3]);
328 obj.time = boost::lexical_cast<double>(results[i][4]);
329 obj.sequence = boost::lexical_cast<double>(results[i][5]);
331 playbackShared->playbackQueue.push_back(obj);
334 g_timeout_add(0, getNextEvent, playbackShared);
337 void DatabaseSink::initDb()
339 if(shared) delete shared;
342 shared->db->init(databaseName->value<std::string>(), tablename, tablecreate);
345 void DatabaseSink::setDatabaseFileName(string filename)
347 bool isLogging = databaseLogging->value<bool>();
352 vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key, zone, source FROM " + tablename);
354 for(int i=0; i < supportedStr.size(); i++)
356 std::string name = supportedStr[i][0];
358 if(!contains(supported(), name))
360 std::string zoneStr = supportedStr[i][1];
361 std::string sourceStr = supportedStr[i][2];
363 DebugOut() << "adding property " << name << " in zone: " << zoneStr << "for source: " << sourceStr << endl;
365 Zone::Type zone = boost::lexical_cast<Zone::Type>(zoneStr);
366 auto property = addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); });
367 property->sourceUuid = sourceStr;
377 routingEngine->updateSupported(supported(), PropertyList(), &source);
380 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
382 VehicleProperty::Property property = value->name;
384 DebugOut() << "Received property change for " << property << endl;
389 if(!contains(supported(), property))
391 addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);});
392 routingEngine->updateSupported(supported(), PropertyList(), &source);
397 obj.value = value->toString();
398 obj.source = value->sourceUuid;
399 obj.time = value->timestamp;
400 obj.sequence = value->sequence;
401 obj.zone = value->zone;
403 shared->queue.append(obj);
407 const std::string DatabaseSink::uuid() const
409 return "9f88156e-cb92-4472-8775-9c08addf50d3";
412 void DatabaseSink::init()
414 if(configuration.find("databaseFile") != configuration.end())
416 databaseName->setValue(configuration["databaseFile"]);
417 setDatabaseFileName(configuration["databaseFile"]);
420 DebugOut() << "databaseLogging: " << databaseLogging->value<bool>() << endl;
422 routingEngine->updateSupported(supported(), PropertyList(), &source);
425 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
427 BaseDB * db = new BaseDB();
428 db->init(databaseName->value<std::string>(), tablename, tablecreate);
433 query<<"SELECT * from "<<tablename<<" WHERE (";
435 for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
437 DebugOut() << "prop: " << (*itr) << endl;
438 if(itr != reply->properties.begin())
441 query<<"key='"<<(*itr)<<"'";
448 query<<" AND time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
451 if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
453 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
456 if(reply->sourceUuid != "")
457 query<<" AND source='"<<reply->sourceUuid<<"'";
459 query<<" AND zone="<<reply->zone;
461 std::vector<std::vector<string>> data = db->select(query.str());
463 DebugOut()<<"Dataset size "<<data.size()<<endl;
467 reply->success = false;
468 reply->error = AsyncPropertyReply::InvalidOperation;
472 reply->success = true;
475 for(auto i=0; i<data.size(); i++)
477 if(data[i].size() != 7)
481 dbobj.key = data[i][0];
482 dbobj.value = data[i][1];
483 dbobj.source = data[i][2];
484 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
485 dbobj.time = boost::lexical_cast<double>(data[i][4]);
486 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
487 dbobj.tripId = data[i][6];
489 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
492 property->timestamp = dbobj.time;
493 property->sequence = dbobj.sequence;
495 reply->values.push_back(property);
500 reply->completed(reply);
505 AsyncPropertyReply *DatabaseSink::setProperty(const AsyncSetPropertyRequest &request)
507 AsyncPropertyReply* reply = AmbPluginImpl::setProperty(request);
509 if(request.property == DatabaseLogging)
511 if(databaseLogging->value<bool>())
520 else if(request.property == DatabaseFile)
522 setDatabaseFileName(databaseName->value<std::string>());
524 else if( request.property == DatabasePlayback)
526 if(playback->value<bool>())
533 playbackShared->stop = true;
542 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
547 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )