plugins/database/databasesink.cpp

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