plugins/database/databasesink.cpp

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