plugins/database/databasesink.cpp

   1 #include "databasesink.h"
   2 #include "abstractroutingengine.h"
   3
   4 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
   5 {
   6         return new DatabaseSinkManager(routingengine, config);
   7 }
   8
   9 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config)
  10         :AbstractSource(engine,config)
  11 {
  12         databaseName = "storage";
  13         tablename = "data";
  14         tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, time REAL, sequence REAL)";
  15         shared = new Shared;
  16         shared->db->init(databaseName, tablename, tablecreate);
  17
  18         engine->subscribeToProperty(VehicleProperty::EngineSpeed, this);
  19         engine->subscribeToProperty(VehicleProperty::VehicleSpeed, this);
  20
  21         PropertyList props;
  22         props.push_back(VehicleProperty::EngineSpeed);
  23         props.push_back(VehicleProperty::VehicleSpeed);
  24
  25         engine->setSupported(supported(),this);
  26
  27         auto cb = [](gpointer data)
  28         {
  29                 Shared *shared = (Shared*)data;
  30
  31                 while(1)
  32                 {
  33                         DBObject* obj = shared->queue.pop();
  34
  35                         if( obj->quit )
  36                         {
  37                                 break;
  38                         }
  39
  40                         DictionaryList<string> dict;
  41
  42                         NameValuePair<string> one("key", obj->key);
  43                         NameValuePair<string> two("value", obj->value);
  44                         NameValuePair<string> three("source", obj->source);
  45                         NameValuePair<string> four("time", boost::lexical_cast<string>(obj->time));
  46                         NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj->sequence));
  47
  48                         dict.push_back(one);
  49                         dict.push_back(two);
  50                         dict.push_back(three);
  51                         dict.push_back(four);
  52                         dict.push_back(five);
  53
  54                         shared->db->insert(dict);
  55                         delete obj;
  56                 }
  57
  58                 void* ret = NULL;
  59                 return ret;
  60         };
  61
  62         thread = g_thread_new("dbthread", cb, shared);
  63
  64 }
  65
  66 DatabaseSink::~DatabaseSink()
  67 {
  68         DBObject* obj = new DBObject();
  69         obj->quit = true;
  70
  71         shared->queue.append(obj);
  72
  73         g_thread_join(thread);
  74
  75         delete shared;
  76 }
  77
  78
  79 void DatabaseSink::supportedChanged(PropertyList supportedProperties)
  80 {
  81
  82 }
  83
  84 PropertyList DatabaseSink::supported()
  85 {
  86         PropertyList props;
  87
  88         props.push_back(VehicleProperty::EngineSpeed);
  89         props.push_back(VehicleProperty::VehicleSpeed);
  90
  91         return props;
  92 }
  93
  94 void DatabaseSink::propertyChanged(VehicleProperty::Property property, AbstractPropertyType *value, std::string uuid)
  95 {
  96         DBObject* obj = new DBObject;
  97         obj->key = property;
  98         obj->value = value->toString();
  99         obj->source = uuid;
 100         obj->time = value->timestamp;
 101         obj->sequence = value->sequence;
 102
 103         shared->queue.append(obj);
 104 }
 105
 106
 107 std::string DatabaseSink::uuid()
 108 {
 109         return "9f88156e-cb92-4472-8775-9c08addf50d3";
 110 }
 111
 112 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
 113 {
 114
 115 }
 116
 117 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 118 {
 119         BaseDB * db = new BaseDB();
 120         db->init(databaseName, tablename, tablecreate);
 121
 122         ostringstream query;
 123         query.precision(15);
 124
 125         query<<"SELECT * from "<<tablename<<" WHERE ";
 126
 127         if(reply->timeBegin && reply->timeEnd)
 128         {
 129                 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
 130         }
 131
 132         if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
 133         {
 134                 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
 135         }
 136
 137         std::vector<std::vector<string>> data = db->select(query.str());
 138
 139         std::list<AbstractPropertyType*> cleanup;
 140
 141         for(auto i=0;i<data.size();i++)
 142         {
 143                 if(data[i].size() != 5)
 144                         continue;
 145
 146                 DBObject dbobj;
 147                 dbobj.key = data[i][0];
 148                 dbobj.value = data[i][1];
 149                 dbobj.source = data[i][2];
 150                 dbobj.time = boost::lexical_cast<double>(data[i][3]);
 151                 dbobj.sequence = boost::lexical_cast<double>(data[i][4]);
 152
 153                 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key,dbobj.value);
 154                 if(property)
 155                 {
 156                         property->timestamp = dbobj.time;
 157                         property->sequence = dbobj.sequence;
 158
 159                         reply->values.push_back(property);
 160                         cleanup.push_back(property);
 161                 }
 162         }
 163
 164         reply->success = true;
 165         reply->completed(reply);
 166
 167         /// reply is owned by the requester of this call.  we own the data:
 168         for(auto itr = cleanup.begin(); itr != cleanup.end(); itr++)
 169         {
 170                 delete *itr;
 171         }
 172
 173         delete db;
 174 }
 175
 176 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
 177 {
 178         AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 179         reply->success = false;
 180         return reply;
 181 }
 182
 183 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
 184 {
 185
 186 }
 187
 188 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )
 189 {
 190 }