X-Git-Url: http://review.tizen.org/git/?a=blobdiff_plain;f=plugins%2Fdatabase%2Fdatabasesink.cpp;h=8b3b0c6ddac7477ed08cfb0603061555b5cce48d;hb=c29570fe11555e886f92215b4338de8fca527faa;hp=423e0435e853b3672299d9713299086790b93add;hpb=159303bc9d81e4d1f9a74ef897815055acfd9e68;p=profile%2Fivi%2Fautomotive-message-broker.git diff --git a/plugins/database/databasesink.cpp b/plugins/database/databasesink.cpp index 423e043..8b3b0c6 100644 --- a/plugins/database/databasesink.cpp +++ b/plugins/database/databasesink.cpp @@ -1,48 +1,61 @@ #include "databasesink.h" #include "abstractroutingengine.h" #include "listplusplus.h" +#include "superptr.hpp" +#include "uuidhelper.h" +#include "ambplugin.h" + +#include int bufferLength = 100; +int timeout=1000; -extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map config) +extern "C" void create(AbstractRoutingEngine* routingengine, map config) { - return new DatabaseSinkManager(routingengine, config); + auto plugin = new AmbPlugin(routingengine, config); + plugin->init(); } -void * cbFunc(gpointer data) +static void * cbFunc(Shared* shared) { - Shared *shared = static_cast(data); - if(!shared) { throw std::runtime_error("Could not cast shared object."); } + ///new tripID: + shared->tripId = amb::createUuid(); + vector > insertList; while(1) { - DBObject* obj = shared->queue.pop(); + usleep(timeout*1000); - if( obj->quit ) + DBObject obj = shared->queue.pop(); + + if( obj.quit ) { - delete obj; break; } DictionaryList dict; - NameValuePair one("key", obj->key); - NameValuePair two("value", obj->value); - NameValuePair three("source", obj->source); - NameValuePair four("time", boost::lexical_cast(obj->time)); - NameValuePair five("sequence", boost::lexical_cast(obj->sequence)); + NameValuePair one("key", obj.key); + NameValuePair two("value", obj.value); + NameValuePair three("source", obj.source); + NameValuePair zone("zone", boost::lexical_cast(obj.zone)); + NameValuePair four("time", boost::lexical_cast(obj.time)); + NameValuePair five("sequence", boost::lexical_cast(obj.sequence)); + NameValuePair six("tripId", boost::lexical_cast(shared->tripId)); dict.push_back(one); dict.push_back(two); dict.push_back(three); + dict.push_back(zone); dict.push_back(four); dict.push_back(five); + dict.push_back(six); insertList.push_back(dict); @@ -57,7 +70,7 @@ void * cbFunc(gpointer data) shared->db->exec("END TRANSACTION"); insertList.clear(); } - delete obj; + //delete obj; } /// final flush of whatever is still in the queue: @@ -90,21 +103,24 @@ int getNextEvent(gpointer data) return 0; } - DBObject* obj = *itr; + DBObject obj = *itr; - AbstractPropertyType* value = VehicleProperty::getPropertyTypeForPropertyNameValue(obj->key,obj->value); + auto value = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key, obj.value)); if(value) { - pbshared->routingEngine->updateProperty(obj->key, value, pbshared->uuid); - value->timestamp = obj->time; - //value->sequence = obj->sequence; + value->priority = AbstractPropertyType::Instant; + value->timestamp = obj.time; + value->sequence = obj.sequence; + value->sourceUuid = obj.source; + value->zone = obj.zone; + pbshared->routingEngine->updateProperty(value.get(), pbshared->uuid); } if(++itr != pbshared->playbackQueue.end()) { - DBObject *o2 = *itr; - double t = o2->time - obj->time; + DBObject o2 = *itr; + double t = o2.time - obj.time; if(t > 0) g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared); @@ -114,26 +130,31 @@ int getNextEvent(gpointer data) pbshared->playbackQueue.remove(obj); DebugOut()<<"playback Queue size: "<playbackQueue.size()< config) - :AbstractSource(engine,config),thread(NULL),shared(NULL),playback(false),playbackShared(NULL), playbackMultiplier(1) +DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map config, AbstractSource &parent) + :AmbPluginImpl(engine, config, parent), shared(nullptr), playback(false), playbackShared(nullptr), playbackMultiplier(1) { - databaseName = "storage"; tablename = "data"; - tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, time REAL, sequence REAL)"; + tablecreate = database_table_create; - if(config.find("databaseFile") != config.end()) + if(config.find("bufferLength") != config.end()) { - setDatabaseFileName(config["databaseFile"]); + bufferLength = boost::lexical_cast(config["bufferLength"]); } - if(config.find("bufferLength") != config.end()) + if(config.find("frequency") != config.end()) { - bufferLength = atoi(config["bufferLength"].c_str()); + try + { + int t = boost::lexical_cast(config["frequency"]); + timeout = 1000 / t; + }catch(...) + { + DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<subscribeToProperty(*itr,this); + engine->subscribeToProperty(itr, &parent); } - mSupported.push_back(DatabaseFile); - mSupported.push_back(DatabaseLogging); - mSupported.push_back(DatabasePlayback); - - routingEngine->setSupported(supported(), this); + databaseName = addPropertySupport(Zone::None, [](){ return new DatabaseFileType("storage"); }); + playback = addPropertySupport(Zone::None, [](){ return new DatabasePlaybackType(false); }); + databaseLogging = addPropertySupport(Zone::None, [](){ return new DatabaseLoggingType(false); }); if(config.find("startOnLoad")!= config.end()) { - setLogging(true); + databaseLogging->setValue(config["startOnLoad"] == "true"); } if(config.find("playbackMultiplier")!= config.end()) @@ -164,24 +183,15 @@ DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, mapsetValue(config["playbackOnLoad"] == "true"); } - - } DatabaseSink::~DatabaseSink() { if(shared) { - DBObject* obj = new DBObject(); - obj->quit = true; - - shared->queue.append(obj); - - g_thread_join(thread); - g_thread_unref(thread); - delete shared; + stopDb(); } if(playbackShared) @@ -191,22 +201,11 @@ DatabaseSink::~DatabaseSink() } -void DatabaseSink::supportedChanged(PropertyList supportedProperties) +void DatabaseSink::supportedChanged(const PropertyList &supportedProperties) { } -PropertyList DatabaseSink::supported() -{ - return mSupported; -} - -PropertyInfo DatabaseSink::getPropertyInfo(VehicleProperty::Property property) -{ - /// TODO: Compute update frequency for properties in the database - return PropertyInfo(); -} - void DatabaseSink::parseConfig() { json_object *rootobject; @@ -224,14 +223,14 @@ void DatabaseSink::parseConfig() { //Should handle the extra data here sometime... } - + json_object *propobject = json_object_object_get(rootobject,"properties"); - + g_assert(json_object_get_type(propobject) == json_type_array); array_list *proplist = json_object_get_array(propobject); - - for(int i=0; i < array_list_length(proplist); i++) + + for(int i=0; i < array_list_length(proplist); i++) { json_object *idxobj = (json_object*)array_list_get_idx(proplist,i); std::string prop = json_object_get_string(idxobj); @@ -240,7 +239,7 @@ void DatabaseSink::parseConfig() DebugOut()<<"DatabaseSink logging: "<quit = true; + DBObject obj; + obj.quit = true; shared->queue.append(obj); - g_thread_join(thread); + if(thread && thread->joinable()) + thread->join(); delete shared; shared = NULL; @@ -261,7 +261,7 @@ void DatabaseSink::stopDb() void DatabaseSink::startDb() { - if(playback) + if(playback->value()) { DebugOut(0)<<"ERROR: tried to start logging during playback. Only logging or playback can be used at one time"<joinable()) + thread->detach(); + + thread = amb::make_unique(new std::thread(cbFunc, shared)); } void DatabaseSink::startPlayback() { - if(playback) + if(playback->value()) return; - playback = true; + playback->setValue(true); initDb(); @@ -309,21 +312,19 @@ void DatabaseSink::startPlayback() throw std::runtime_error("column mismatch in query"); } - DBObject* obj = new DBObject(); - - obj->key = results[i][0]; - obj->value = results[i][1]; - obj->source = results[i][2]; - obj->time = boost::lexical_cast(results[i][3]); + DBObject obj; - /// TODO: figure out why sequence is broken: - -// obj->sequence = boost::lexical_cast(results[i][4]); + obj.key = results[i][0]; + obj.value = results[i][1]; + obj.source = results[i][2]; + obj.zone = boost::lexical_cast(results[i][3]); + obj.time = boost::lexical_cast(results[i][4]); + obj.sequence = boost::lexical_cast(results[i][5]); playbackShared->playbackQueue.push_back(obj); } - g_timeout_add(0,getNextEvent,playbackShared); + g_timeout_add(0, getNextEvent, playbackShared); } void DatabaseSink::initDb() @@ -331,125 +332,103 @@ void DatabaseSink::initDb() if(shared) delete shared; shared = new Shared; - shared->db->init(databaseName, tablename, tablecreate); -} - -void DatabaseSink::setPlayback(bool v) -{ - AsyncSetPropertyRequest request; - request.property = DatabasePlayback; - request.value = new DatabasePlaybackType(v); - - setProperty(request); -} - -void DatabaseSink::setLogging(bool b) -{ - AsyncSetPropertyRequest request; - request.property = DatabaseLogging; - request.value = new DatabaseLoggingType(b); - - setProperty(request); + shared->db->init(databaseName->value(), tablename, tablecreate); } void DatabaseSink::setDatabaseFileName(string filename) { - databaseName = filename; - initDb(); - vector > supportedStr = shared->db->select("SELECT DISTINCT key FROM "+tablename); + vector > supportedStr = shared->db->select("SELECT DISTINCT key, zone, source FROM " + tablename); for(int i=0; i < supportedStr.size(); i++) { - if(!ListPlusPlus(&mSupported).contains(supportedStr[i][0])) - mSupported.push_back(supportedStr[i][0]); + std::string name = supportedStr[i][0]; + + if(!contains(supported(), name)) + { + std::string zoneStr = supportedStr[i][1]; + std::string sourceStr = supportedStr[i][2]; + + DebugOut() << "adding property " << name << " in zone: " << zoneStr << "for source: " << sourceStr << endl; + + Zone::Type zone = boost::lexical_cast(zoneStr); + auto property = addPropertySupport(zone, [name]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(name); }); + property->sourceUuid = sourceStr; + } } delete shared; shared = NULL; - routingEngine->setSupported(mSupported, this); + routingEngine->updateSupported(supported(), PropertyList(), &source); } -void DatabaseSink::propertyChanged(VehicleProperty::Property property, AbstractPropertyType *value, std::string uuid) +void DatabaseSink::propertyChanged(AbstractPropertyType *value) { + VehicleProperty::Property property = value->name; + if(!shared) return; - if(!ListPlusPlus(&mSupported).contains(property)) + if(!contains(supported(), property)) { - mSupported.push_back(property); - routingEngine->setSupported(mSupported, this); + addPropertySupport(value->zone, [property]() { return VehicleProperty::getPropertyTypeForPropertyNameValue(property);}); + routingEngine->updateSupported(supported(), PropertyList(), &source); } - DBObject* obj = new DBObject; - obj->key = property; - obj->value = value->toString(); - obj->source = uuid; - obj->time = value->timestamp; - obj->sequence = value->sequence; + DBObject obj; + obj.key = property; + obj.value = value->toString(); + obj.source = value->sourceUuid; + obj.time = value->timestamp; + obj.sequence = value->sequence; + obj.zone = value->zone; shared->queue.append(obj); } -std::string DatabaseSink::uuid() +const std::string DatabaseSink::uuid() const { return "9f88156e-cb92-4472-8775-9c08addf50d3"; } -void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply) +void DatabaseSink::init() { - reply->success = false; - - if(reply->property == DatabaseFile) - { - DatabaseFileType temp(databaseName); - reply->value = &temp; - - reply->success = true; - reply->completed(reply); - - return; - } - else if(reply->property == DatabaseLogging) - { - DatabaseLoggingType temp = shared; - - reply->value = &temp; - reply->success = true; - reply->completed(reply); - - return; - } - - else if(reply->property == DatabasePlayback) + if(configuration.find("databaseFile") != configuration.end()) { - DatabasePlaybackType temp = playback; - reply->value = &temp; - reply->success = true; - reply->completed(reply); - - return; + databaseName->setValue(configuration["databaseFile"]); + setDatabaseFileName(configuration["databaseFile"]); } - reply->completed(reply); + routingEngine->updateSupported(supported(), PropertyList(), &source); } void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply) { BaseDB * db = new BaseDB(); - db->init(databaseName, tablename, tablecreate); + db->init(databaseName->value(), tablename, tablecreate); ostringstream query; query.precision(15); - query<<"SELECT * from "<timeBegin && reply->timeEnd) + for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++) { - query<<" time BETWEEN "<timeBegin<<" AND "<timeEnd; + DebugOut() << "prop: " << (*itr) << endl; + if(itr != reply->properties.begin()) + query<<" OR "; + + query<<"key='"<<(*itr)<<"'"; + } + + query<<")"; + + if(reply->timeEnd) + { + query<<" AND time BETWEEN "<timeBegin<<" AND "<timeEnd; } if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0) @@ -457,21 +436,38 @@ void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply) query<<" AND sequence BETWEEN "<sequenceBegin<<" AND "<sequenceEnd; } + if(reply->sourceUuid != "") + query<<" AND source='"<sourceUuid<<"'"; + + query<<" AND zone="<zone; + std::vector> data = db->select(query.str()); - std::list cleanup; + DebugOut()<<"Dataset size "<success = false; + reply->error = AsyncPropertyReply::InvalidOperation; + } + else { - if(data[i].size() != 5) + reply->success = true; + } + + for(auto i=0; i(data[i][3]); - dbobj.sequence = boost::lexical_cast(data[i][4]); + dbobj.zone = boost::lexical_cast(data[i][3]); + dbobj.time = boost::lexical_cast(data[i][4]); + dbobj.sequence = boost::lexical_cast(data[i][5]); + dbobj.tripId = data[i][6]; AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value); if(property) @@ -480,11 +476,10 @@ void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply) property->sequence = dbobj.sequence; reply->values.push_back(property); - cleanup.push_back(property); } } - reply->success = true; + reply->completed(reply); delete db; @@ -492,50 +487,28 @@ void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply) AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request) { - AsyncPropertyReply* reply = new AsyncPropertyReply(request); - reply->success = false; + AsyncPropertyReply* reply = AmbPluginImpl::setProperty(request); if(request.property == DatabaseLogging) { if(request.value->value()) { - setPlayback(false); startDb(); - reply->success = true; - DatabaseLoggingType temp(true); - routingEngine->updateProperty(DatabaseLogging,&temp,uuid()); } else { stopDb(); - reply->success = true; - DatabaseLoggingType temp(false); - routingEngine->updateProperty(DatabaseLogging,&temp,uuid()); } } - else if(request.property == DatabaseFile) { - std::string fname = request.value->toString(); - - databaseName = fname; - - DatabaseFileType temp(databaseName); - - routingEngine->updateProperty(DatabaseFile,&temp,uuid()); - - reply->success = true; + setDatabaseFileName(databaseName->value()); } else if( request.property == DatabasePlayback) { if(request.value->value()) { - setLogging(false); startPlayback(); - - DatabasePlaybackType temp(playback); - - routingEngine->updateProperty(DatabasePlayback,&temp,uuid()); } else { @@ -543,13 +516,7 @@ AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request) playbackShared->stop = true; playback = false; - - DatabasePlaybackType temp(playback); - - routingEngine->updateProperty(DatabasePlayback, &temp, uuid()); } - - reply->success = true; } return reply; @@ -563,3 +530,4 @@ void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property ) void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property ) { } +