plugins/database/databasesink.cpp

   1 #include "databasesink.h"
   2 #include "abstractroutingengine.h"
   3 #include "listplusplus.h"
   4
   5 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
   6 {
   7         return new DatabaseSinkManager(routingengine, config);
   8 }
   9
  10 void * cbFunc(gpointer data)
  11 {
  12         Shared *shared = static_cast<Shared*>(data);
  13
  14         if(!shared)
  15         {
  16                 throw std::runtime_error("Could not cast shared object.");
  17         }
  18
  19         vector<DictionaryList<string> > insertList;
  20
  21         while(1)
  22         {
  23                 DBObject* obj = shared->queue.pop();
  24
  25                 if( obj->quit )
  26                 {
  27                         delete obj;
  28                         break;
  29                 }
  30
  31                 DictionaryList<string> dict;
  32
  33                 NameValuePair<string> one("key", obj->key);
  34                 NameValuePair<string> two("value", obj->value);
  35                 NameValuePair<string> three("source", obj->source);
  36                 NameValuePair<string> four("time", boost::lexical_cast<string>(obj->time));
  37                 NameValuePair<string> five("sequence", boost::lexical_cast<string>(obj->sequence));
  38
  39                 dict.push_back(one);
  40                 dict.push_back(two);
  41                 dict.push_back(three);
  42                 dict.push_back(four);
  43                 dict.push_back(five);
  44
  45                 insertList.push_back(dict);
  46
  47                 if(insertList.size() > 100)
  48                 {
  49                         shared->db->exec("BEGIN IMMEDIATE TRANSACTION");
  50                         for(int i=0; i< insertList.size(); i++)
  51                         {
  52                                 DictionaryList<string> d = insertList[i];
  53                                 shared->db->insert(d);
  54                         }
  55                         shared->db->exec("END TRANSACTION");
  56                         insertList.clear();
  57                 }
  58                 delete obj;
  59         }
  60
  61         return NULL;
  62 }
  63
  64 int getNextEvent(gpointer data)
  65 {
  66         PlaybackShared* pbshared = static_cast<PlaybackShared*>(data);
  67
  68         if(!pbshared)
  69                 throw std::runtime_error("failed to cast PlaybackShared object");
  70
  71         auto itr = pbshared->playbackQueue.begin();
  72
  73         if(itr == pbshared->playbackQueue.end())
  74         {
  75                 return 0;
  76         }
  77
  78         DBObject* obj = *itr;
  79
  80         AbstractPropertyType* value = VehicleProperty::getPropertyTypeForPropertyNameValue(obj->key,obj->value);
  81
  82         if(value)
  83         {
  84                 pbshared->routingEngine->updateProperty(obj->key, value, pbshared->uuid);
  85                 value->timestamp = obj->time;
  86                 //value->sequence = obj->sequence;
  87         }
  88
  89         if(++itr != pbshared->playbackQueue.end())
  90         {
  91                 DBObject *o2 = *itr;
  92                 double t = o2->time - obj->time;
  93
  94                 if(t > 0)
  95                         g_timeout_add((t*1000) / pbshared->playBackMultiplier, getNextEvent, pbshared);
  96                 else
  97                         g_timeout_add(t, getNextEvent, pbshared);
  98         }
  99
 100         pbshared->playbackQueue.remove(obj);
 101         delete obj;
 102
 103         return 0;
 104 }
 105
 106 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config)
 107         :AbstractSource(engine,config),thread(NULL),shared(NULL),playback(false),playbackShared(NULL), playbackMultiplier(1)
 108 {
 109         databaseName = "storage";
 110         tablename = "data";
 111         tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, time REAL, sequence REAL)";
 112
 113         if(config.find("databaseFile") != config.end())
 114         {
 115                 databaseName = config["databaseFile"];
 116         }
 117
 118         if(config.find("properties") != config.end())
 119         {
 120                 parseConfig();
 121         }
 122
 123         for(auto itr=propertiesToSubscribeTo.begin();itr!=propertiesToSubscribeTo.end();itr++)
 124         {
 125                 engine->subscribeToProperty(*itr,this);
 126         }
 127
 128         mSupported.push_back(DatabaseFileProperty);
 129         mSupported.push_back(DatabaseLoggingProperty);
 130         mSupported.push_back(DatabasePlaybackProperty);
 131
 132
 133         initDb();
 134
 135         /// get supported:
 136
 137         vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key FROM "+tablename);
 138
 139         for(int i=0; i < supportedStr.size(); i++)
 140         {
 141                 if(!ListPlusPlus<VehicleProperty::Property>(&mSupported).contains(supportedStr[i][0]))
 142                         mSupported.push_back(supportedStr[i][0]);
 143         }
 144
 145         routingEngine->setSupported(supported(), this);
 146
 147         if(config.find("startOnLoad")!= config.end())
 148         {
 149                 AsyncSetPropertyRequest request;
 150                 request.property = DatabaseLoggingProperty;
 151                 request.value = new BasicPropertyType<bool>(true);
 152
 153                 setProperty(request);
 154         }
 155
 156         if(config.find("playbackMultiplier")!= config.end())
 157         {
 158                 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
 159         }
 160
 161         if(config.find("playbackOnLoad")!= config.end())
 162         {
 163                 startPlayback();
 164         }
 165
 166
 167 }
 168
 169 DatabaseSink::~DatabaseSink()
 170 {
 171         if(shared)
 172         {
 173                 DBObject* obj = new DBObject();
 174                 obj->quit = true;
 175
 176                 shared->queue.append(obj);
 177
 178                 g_thread_join(thread);
 179                 g_thread_unref(thread);
 180                 delete shared;
 181         }
 182
 183         if(playbackShared)
 184         {
 185                 delete playbackShared;
 186         }
 187 }
 188
 189
 190 void DatabaseSink::supportedChanged(PropertyList supportedProperties)
 191 {
 192
 193 }
 194
 195 PropertyList DatabaseSink::supported()
 196 {
 197         return mSupported;
 198 }
 199
 200 void DatabaseSink::parseConfig()
 201 {
 202         json_object *rootobject;
 203         json_tokener *tokener = json_tokener_new();
 204         enum json_tokener_error err;
 205         do
 206         {
 207                 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
 208         } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
 209         if (err != json_tokener_success)
 210         {
 211                 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
 212         }
 213         if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
 214         {
 215                 //Should handle the extra data here sometime...
 216         }
 217
 218         json_object *propobject = json_object_object_get(rootobject,"properties");
 219
 220         g_assert(json_object_get_type(propobject) == json_type_array);
 221
 222         array_list *proplist = json_object_get_array(propobject);
 223
 224         for(int i=0; i < array_list_length(proplist); i++)
 225         {
 226                 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
 227                 std::string prop = json_object_get_string(idxobj);
 228                 propertiesToSubscribeTo.push_back(prop);
 229
 230                 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
 231         }
 232
 233         json_object_put(propobject);
 234         json_object_put(rootobject);
 235 }
 236
 237 void DatabaseSink::stopDb()
 238 {
 239         if(!shared)
 240                 return;
 241
 242         DBObject *obj = new DBObject();
 243         obj->quit = true;
 244         shared->queue.append(obj);
 245
 246         g_thread_join(thread);
 247
 248         delete shared;
 249         shared = NULL;
 250 }
 251
 252 void DatabaseSink::startDb()
 253 {
 254         if(playback)
 255         {
 256                 DebugOut(0)<<"ERROR: tried to start logging during playback.  Only logging or playback can be used at one time"<<endl;
 257                 return;
 258         }
 259
 260         if(shared)
 261         {
 262                 DebugOut(0)<<"WARNING: logging already started.  doing nothing."<<endl;
 263                 return;
 264         }
 265
 266         initDb();
 267
 268         thread = g_thread_new("dbthread", cbFunc, shared);
 269 }
 270
 271 void DatabaseSink::startPlayback()
 272 {
 273         if(playback)
 274                 return;
 275
 276         playback = true;
 277
 278         initDb();
 279
 280         /// populate playback queue:
 281
 282         vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
 283
 284         if(playbackShared)
 285         {
 286                 delete playbackShared;
 287         }
 288
 289         playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
 290
 291         for(int i=0;i<results.size();i++)
 292         {
 293                 if(results[i].size() < 5)
 294                 {
 295                         throw std::runtime_error("column mismatch in query");
 296                 }
 297
 298                 DBObject* obj = new DBObject();
 299
 300                 obj->key = results[i][0];
 301                 obj->value = results[i][1];
 302                 obj->source = results[i][2];
 303                 obj->time = boost::lexical_cast<double>(results[i][3]);
 304 //              obj->sequence = boost::lexical_cast<int>(results[i][4]);
 305
 306                 playbackShared->playbackQueue.push_back(obj);
 307         }
 308
 309         g_timeout_add(0,getNextEvent,playbackShared);
 310 }
 311
 312 void DatabaseSink::initDb()
 313 {
 314         if(shared) delete shared;
 315
 316         shared = new Shared;
 317         shared->db->init(databaseName, tablename, tablecreate);
 318 }
 319
 320 void DatabaseSink::propertyChanged(VehicleProperty::Property property, AbstractPropertyType *value, std::string uuid)
 321 {
 322         if(!shared)
 323                 return;
 324
 325         DBObject* obj = new DBObject;
 326         obj->key = property;
 327         obj->value = value->toString();
 328         obj->source = uuid;
 329         obj->time = value->timestamp;
 330         obj->sequence = value->sequence;
 331
 332         shared->queue.append(obj);
 333 }
 334
 335
 336 std::string DatabaseSink::uuid()
 337 {
 338         return "9f88156e-cb92-4472-8775-9c08addf50d3";
 339 }
 340
 341 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
 342 {
 343         reply->success = false;
 344
 345         if(reply->property == DatabaseFileProperty)
 346         {
 347                 StringPropertyType temp(databaseName);
 348                 reply->value = &temp;
 349
 350                 reply->success = true;
 351                 reply->completed(reply);
 352
 353                 return;
 354         }
 355         else if(reply->property == DatabaseLoggingProperty)
 356         {
 357                 BasicPropertyType<bool> temp = shared;
 358
 359                 reply->value = &temp;
 360                 reply->success = true;
 361                 reply->completed(reply);
 362
 363                 return;
 364         }
 365
 366         else if(reply->property == DatabasePlaybackProperty)
 367         {
 368                 BasicPropertyType<bool> temp = playback;
 369                 reply->value = &temp;
 370                 reply->success = true;
 371                 reply->completed(reply);
 372
 373                 return;
 374         }
 375
 376         reply->completed(reply);
 377 }
 378
 379 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 380 {
 381         BaseDB * db = new BaseDB();
 382         db->init(databaseName, tablename, tablecreate);
 383
 384         ostringstream query;
 385         query.precision(15);
 386
 387         query<<"SELECT * from "<<tablename<<" WHERE ";
 388
 389         if(reply->timeBegin && reply->timeEnd)
 390         {
 391                 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
 392         }
 393
 394         if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
 395         {
 396                 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
 397         }
 398
 399         std::vector<std::vector<string>> data = db->select(query.str());
 400
 401         std::list<AbstractPropertyType*> cleanup;
 402
 403         for(auto i=0;i<data.size();i++)
 404         {
 405                 if(data[i].size() != 5)
 406                         continue;
 407
 408                 DBObject dbobj;
 409                 dbobj.key = data[i][0];
 410                 dbobj.value = data[i][1];
 411                 dbobj.source = data[i][2];
 412                 dbobj.time = boost::lexical_cast<double>(data[i][3]);
 413                 dbobj.sequence = boost::lexical_cast<double>(data[i][4]);
 414
 415                 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
 416                 if(property)
 417                 {
 418                         property->timestamp = dbobj.time;
 419                         property->sequence = dbobj.sequence;
 420
 421                         reply->values.push_back(property);
 422                         cleanup.push_back(property);
 423                 }
 424         }
 425
 426         reply->success = true;
 427         reply->completed(reply);
 428
 429         delete db;
 430 }
 431
 432 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
 433 {
 434         AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 435         reply->success = false;
 436
 437         if(request.property == DatabaseLoggingProperty)
 438         {
 439                 if(request.value->value<bool>())
 440                 {
 441                         ///TODO: start or stop logging thread
 442                         startDb();
 443                         reply->success = true;
 444                         BasicPropertyType<bool> temp(true);
 445                         routingEngine->updateProperty(DatabaseLoggingProperty,&temp,uuid());
 446                 }
 447                 else
 448                 {
 449                         stopDb();
 450                         reply->success = true;
 451                         BasicPropertyType<bool> temp(false);
 452                         routingEngine->updateProperty(DatabaseLoggingProperty,&temp,uuid());
 453                 }
 454         }
 455
 456         else if(request.property == DatabaseFileProperty)
 457         {
 458                 std::string fname = request.value->toString();
 459
 460                 databaseName = fname;
 461
 462                 StringPropertyType temp(databaseName);
 463
 464                 routingEngine->updateProperty(DatabaseFileProperty,&temp,uuid());
 465
 466                 reply->success = true;
 467         }
 468         else if( request.property == DatabasePlaybackProperty)
 469         {
 470                 if(request.value->value<bool>())
 471                 {
 472                         startPlayback();
 473
 474                         BasicPropertyType<bool> temp(true);
 475
 476                         routingEngine->updateProperty(DatabasePlaybackProperty,&temp,uuid());
 477                 }
 478                 else
 479                 {
 480                         /// TODO: stop playback
 481
 482                         BasicPropertyType<bool> temp(true);
 483
 484                         routingEngine->updateProperty(DatabasePlaybackProperty,&temp,uuid());
 485                 }
 486
 487                 reply->success = true;
 488         }
 489
 490         return reply;
 491 }
 492
 493 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
 494 {
 495
 496 }
 497
 498 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )
 499 {
 500 }