plugins/database/databasesink.cpp

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