added frequency option to database plugin. fixed up zone a bit.

[profile/ivi/automotive-message-broker.git] / plugins / database / databasesink.cpp

diff --git a/plugins/database/databasesink.cpp b/plugins/database/databasesink.cpp
index f4790be..bf388eb 100644 (file)
--- a/plugins/database/databasesink.cpp
+++ b/plugins/database/databasesink.cpp
@@ -3,6 +3,7 @@
 #include "listplusplus.h"
 
 int bufferLength = 100;
+int timeout=1000;
 
 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
 {
@@ -22,22 +23,23 @@ void * cbFunc(gpointer data)
 
        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<string> dict;
 
-               NameValuePair<string> one("key", obj->key);
-               NameValuePair<string> two("value", obj->value);
-               NameValuePair<string> three("source", obj->source);
-               NameValuePair<string> zone("zone", boost::lexical_cast<string>(obj->zone));
-               NameValuePair<string> four("time", boost::lexical_cast<string>(obj->time));
-               NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj->sequence));
+               NameValuePair<string> one("key", obj.key);
+               NameValuePair<string> two("value", obj.value);
+               NameValuePair<string> three("source", obj.source);
+               NameValuePair<string> zone("zone", boost::lexical_cast<string>(obj.zone));
+               NameValuePair<string> four("time", boost::lexical_cast<string>(obj.time));
+               NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj.sequence));
 
                dict.push_back(one);
                dict.push_back(two);
@@ -59,7 +61,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:
@@ -92,22 +94,23 @@ int getNextEvent(gpointer data)
                return 0;
        }
 
-       DBObject* obj = *itr;
+       DBObject obj = *itr;
 
-       AbstractPropertyType* value = VehicleProperty::getPropertyTypeForPropertyNameValue(obj->key,obj->value);
+       AbstractPropertyType* value = VehicleProperty::getPropertyTypeForPropertyNameValue(obj.key,obj.value);
 
        if(value)
        {
                pbshared->routingEngine->updateProperty(value, pbshared->uuid);
-               value->timestamp = obj->time;
-               value->sequence = obj->sequence;
-               value->sourceUuid = obj->source;
+               value->timestamp = obj.time;
+               value->sequence = obj.sequence;
+               value->sourceUuid = obj.source;
+               value->zone = obj.zone;
        }
 
        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);
@@ -117,7 +120,7 @@ int getNextEvent(gpointer data)
 
        pbshared->playbackQueue.remove(obj);
        DebugOut()<<"playback Queue size: "<<pbshared->playbackQueue.size()<<endl;
-       delete obj;
+       //delete obj;
 
        return 0;
 }
@@ -139,6 +142,20 @@ DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::
                bufferLength = atoi(config["bufferLength"].c_str());
        }
 
+       if(config.find("frequency") != config.end())
+       {
+               try
+               {
+                       int t = boost::lexical_cast<int>(config["frequency"]);
+                       timeout = 1000 / t;
+               }catch(...)
+               {
+                       DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
+               }
+
+
+       }
+
        if(config.find("properties") != config.end())
        {
                parseConfig();
@@ -177,14 +194,7 @@ 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)
@@ -252,8 +262,8 @@ void DatabaseSink::stopDb()
        if(!shared)
                return;
 
-       DBObject *obj = new DBObject();
-       obj->quit = true;
+       DBObject obj;
+       obj.quit = true;
        shared->queue.append(obj);
 
        g_thread_join(thread);
@@ -312,12 +322,12 @@ void DatabaseSink::startPlayback()
                        throw std::runtime_error("column mismatch in query");
                }
 
-               DBObject* obj = new DBObject();
+               DBObject obj;
 
-               obj->key = results[i][0];
-               obj->value = results[i][1];
-               obj->source = results[i][2];
-               obj->time = boost::lexical_cast<double>(results[i][3]);
+               obj.key = results[i][0];
+               obj.value = results[i][1];
+               obj.source = results[i][2];
+               obj.time = boost::lexical_cast<double>(results[i][3]);
 
                /// TODO: figure out why sequence is broken:
 
@@ -390,12 +400,13 @@ void DatabaseSink::propertyChanged(AbstractPropertyType *value)
                routingEngine->setSupported(mSupported, this);
        }
 
-       DBObject* obj = new DBObject;
-       obj->key = property;
-       obj->value = value->toString();
-       obj->source = value->sourceUuid;
-       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);
 }