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         //delete obj;
 126
 127         return 0;
 128 }
 129
 130 DatabaseSink::DatabaseSink(AbstractRoutingEngine *engine, map<std::string, std::string> config)
 131         :AbstractSource(engine,config),thread(NULL),shared(NULL),playback(false),playbackShared(NULL), playbackMultiplier(1)
 132 {
 133         databaseName = "storage";
 134         tablename = "data";
 135         tablecreate = "CREATE TABLE IF NOT EXISTS data (key TEXT, value BLOB, source TEXT, zone REAL, time REAL, sequence REAL)";
 136
 137         if(config.find("databaseFile") != config.end())
 138         {
 139                 setDatabaseFileName(config["databaseFile"]);
 140         }
 141
 142         if(config.find("bufferLength") != config.end())
 143         {
 144                 bufferLength = atoi(config["bufferLength"].c_str());
 145         }
 146
 147         if(config.find("frequency") != config.end())
 148         {
 149                 try
 150                 {
 151                         int t = boost::lexical_cast<int>(config["frequency"]);
 152                         timeout = 1000 / t;
 153                 }catch(...)
 154                 {
 155                         DebugOut(DebugOut::Error)<<"Failed to parse frequency: Invalid value "<<config["frequency"]<<endl;
 156                 }
 157
 158
 159         }
 160
 161         if(config.find("properties") != config.end())
 162         {
 163                 parseConfig();
 164         }
 165
 166         for(auto itr=propertiesToSubscribeTo.begin();itr!=propertiesToSubscribeTo.end();itr++)
 167         {
 168                 engine->subscribeToProperty(*itr,this);
 169         }
 170
 171         mSupported.push_back(DatabaseFile);
 172         mSupported.push_back(DatabaseLogging);
 173         mSupported.push_back(DatabasePlayback);
 174
 175         routingEngine->updateSupported(supported(), PropertyList(), this);
 176
 177         if(config.find("startOnLoad")!= config.end())
 178         {
 179                 setLogging(config["startOnLoad"] == "true");
 180         }
 181
 182         if(config.find("playbackMultiplier")!= config.end())
 183         {
 184                 playbackMultiplier = boost::lexical_cast<uint>(config["playbackMultiplier"]);
 185         }
 186
 187         if(config.find("playbackOnLoad")!= config.end())
 188         {
 189                 setPlayback(config["playbackOnLoad"] == "true");
 190         }
 191
 192
 193 }
 194
 195 DatabaseSink::~DatabaseSink()
 196 {
 197         if(shared)
 198         {
 199                 stopDb();
 200         }
 201
 202         if(playbackShared)
 203         {
 204                 delete playbackShared;
 205         }
 206 }
 207
 208
 209 void DatabaseSink::supportedChanged(const PropertyList &supportedProperties)
 210 {
 211
 212 }
 213
 214 PropertyList DatabaseSink::supported()
 215 {
 216         return mSupported;
 217 }
 218
 219 PropertyInfo DatabaseSink::getPropertyInfo(VehicleProperty::Property property)
 220 {
 221         /// TODO: Compute update frequency for properties in the database
 222         return PropertyInfo::invalid();
 223 }
 224
 225 void DatabaseSink::parseConfig()
 226 {
 227         json_object *rootobject;
 228         json_tokener *tokener = json_tokener_new();
 229         enum json_tokener_error err;
 230         do
 231         {
 232                 rootobject = json_tokener_parse_ex(tokener, configuration["properties"].c_str(),configuration["properties"].size());
 233         } while ((err = json_tokener_get_error(tokener)) == json_tokener_continue);
 234         if (err != json_tokener_success)
 235         {
 236                 fprintf(stderr, "Error: %s\n", json_tokener_error_desc(err));
 237         }
 238         if (tokener->char_offset < configuration["properties"].size()) // XXX shouldn't access internal fields
 239         {
 240                 //Should handle the extra data here sometime...
 241         }
 242
 243         json_object *propobject = json_object_object_get(rootobject,"properties");
 244
 245         g_assert(json_object_get_type(propobject) == json_type_array);
 246
 247         array_list *proplist = json_object_get_array(propobject);
 248
 249         for(int i=0; i < array_list_length(proplist); i++)
 250         {
 251                 json_object *idxobj = (json_object*)array_list_get_idx(proplist,i);
 252                 std::string prop = json_object_get_string(idxobj);
 253                 propertiesToSubscribeTo.push_back(prop);
 254
 255                 DebugOut()<<"DatabaseSink logging: "<<prop<<endl;
 256         }
 257
 258         //json_object_put(propobject);
 259         json_object_put(rootobject);
 260 }
 261
 262 void DatabaseSink::stopDb()
 263 {
 264         if(!shared)
 265                 return;
 266
 267         DBObject obj;
 268         obj.quit = true;
 269         shared->queue.append(obj);
 270
 271         g_thread_join(thread);
 272
 273         delete shared;
 274         shared = NULL;
 275 }
 276
 277 void DatabaseSink::startDb()
 278 {
 279         if(playback.basicValue())
 280         {
 281                 DebugOut(0)<<"ERROR: tried to start logging during playback.  Only logging or playback can be used at one time"<<endl;
 282                 return;
 283         }
 284
 285         if(shared)
 286         {
 287                 DebugOut(0)<<"WARNING: logging already started.  doing nothing."<<endl;
 288                 return;
 289         }
 290
 291         initDb();
 292
 293         thread = g_thread_new("dbthread", cbFunc, shared);
 294 }
 295
 296 void DatabaseSink::startPlayback()
 297 {
 298         if(playback.basicValue())
 299                 return;
 300
 301         playback = true;
 302
 303         initDb();
 304
 305         /// populate playback queue:
 306
 307         vector<vector<string> > results = shared->db->select("SELECT * FROM "+tablename);
 308
 309         /// we are done with shared.  clean up:
 310         delete shared;
 311         shared = NULL;
 312
 313         if(playbackShared)
 314         {
 315                 delete playbackShared;
 316         }
 317
 318         playbackShared = new PlaybackShared(routingEngine, uuid(), playbackMultiplier);
 319
 320         for(int i=0;i<results.size();i++)
 321         {
 322                 if(results[i].size() < 5)
 323                 {
 324                         throw std::runtime_error("column mismatch in query");
 325                 }
 326
 327                 DBObject obj;
 328
 329                 obj.key = results[i][0];
 330                 obj.value = results[i][1];
 331                 obj.source = results[i][2];
 332                 obj.zone = boost::lexical_cast<double>(results[i][3]);
 333                 obj.time = boost::lexical_cast<double>(results[i][4]);
 334                 obj.sequence = boost::lexical_cast<double>(results[i][5]);
 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.value<std::string>(), 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         databaseLogging = b;
 362         AsyncSetPropertyRequest request;
 363         request.property = DatabaseLogging;
 364         request.value = &databaseLogging;
 365
 366         setProperty(request);
 367 }
 368
 369 void DatabaseSink::setDatabaseFileName(string filename)
 370 {
 371         databaseName = filename;
 372
 373         initDb();
 374
 375         vector<vector<string> > supportedStr = shared->db->select("SELECT DISTINCT key FROM "+tablename);
 376
 377         for(int i=0; i < supportedStr.size(); i++)
 378         {
 379                 if(!contains(mSupported, supportedStr[i][0]))
 380                         mSupported.push_back(supportedStr[i][0]);
 381         }
 382
 383         delete shared;
 384         shared = NULL;
 385
 386         routingEngine->updateSupported(mSupported, PropertyList(), this);
 387 }
 388
 389 void DatabaseSink::propertyChanged(AbstractPropertyType *value)
 390 {
 391         VehicleProperty::Property property = value->name;
 392
 393         if(!shared)
 394                 return;
 395
 396         if(!contains(mSupported, property))
 397         {
 398                 mSupported.push_back(property);
 399                 routingEngine->updateSupported(mSupported, PropertyList(), this);
 400         }
 401
 402         DBObject obj;
 403         obj.key = property;
 404         obj.value = value->toString();
 405         obj.source = value->sourceUuid;
 406         obj.time = value->timestamp;
 407         obj.sequence = value->sequence;
 408         obj.zone = value->zone;
 409
 410         shared->queue.append(obj);
 411 }
 412
 413
 414 const string DatabaseSink::uuid()
 415 {
 416         return "9f88156e-cb92-4472-8775-9c08addf50d3";
 417 }
 418
 419 void DatabaseSink::getPropertyAsync(AsyncPropertyReply *reply)
 420 {
 421         reply->success = false;
 422
 423         if(reply->property == DatabaseFile)
 424         {
 425                 DatabaseFileType temp(databaseName);
 426                 reply->value = &temp;
 427
 428                 reply->success = true;
 429                 reply->completed(reply);
 430
 431                 return;
 432         }
 433         else if(reply->property == DatabaseLogging)
 434         {
 435                 databaseLogging = shared != nullptr;
 436
 437                 reply->value = &databaseLogging;
 438                 reply->success = true;
 439                 reply->completed(reply);
 440
 441                 return;
 442         }
 443
 444         else if(reply->property == DatabasePlayback)
 445         {
 446                 DatabasePlaybackType temp = playback;
 447                 reply->value = &temp;
 448                 reply->success = true;
 449                 reply->completed(reply);
 450
 451                 return;
 452         }
 453
 454         reply->completed(reply);
 455 }
 456
 457 void DatabaseSink::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 458 {
 459         BaseDB * db = new BaseDB();
 460         db->init(databaseName.value<std::string>(), tablename, tablecreate);
 461
 462         ostringstream query;
 463         query.precision(15);
 464
 465         query<<"SELECT * from "<<tablename<<" WHERE (";
 466
 467         for(auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
 468         {
 469                 if(itr != reply->properties.begin())
 470                         query<<" OR ";
 471
 472                 query<<"key='"<<(*itr)<<"'";
 473         }
 474
 475         query<<") AND";
 476
 477         if(reply->timeBegin && reply->timeEnd)
 478         {
 479                 query<<" time BETWEEN "<<reply->timeBegin<<" AND "<<reply->timeEnd;
 480         }
 481
 482         if(reply->sequenceBegin >= 0 && reply->sequenceEnd >=0)
 483         {
 484                 query<<" AND sequence BETWEEN "<<reply->sequenceBegin<<" AND "<<reply->sequenceEnd;
 485         }
 486
 487         if(reply->sourceUuid != "")
 488                 query<<" AND source='"<<reply->sourceUuid<<"'";
 489
 490         query<<" AND zone="<<reply->zone;
 491
 492         std::vector<std::vector<string>> data = db->select(query.str());
 493
 494         DebugOut()<<"Dataset size "<<data.size()<<endl;
 495
 496         for(auto i=0;i<data.size();i++)
 497         {
 498                 if(data[i].size() != 6)
 499                         continue;
 500
 501                 DBObject dbobj;
 502                 dbobj.key = data[i][0];
 503                 dbobj.value = data[i][1];
 504                 dbobj.source = data[i][2];
 505                 dbobj.zone = boost::lexical_cast<double>(data[i][3]);
 506                 dbobj.time = boost::lexical_cast<double>(data[i][4]);
 507                 dbobj.sequence = boost::lexical_cast<double>(data[i][5]);
 508
 509                 AbstractPropertyType* property = VehicleProperty::getPropertyTypeForPropertyNameValue(dbobj.key, dbobj.value);
 510                 if(property)
 511                 {
 512                         property->timestamp = dbobj.time;
 513                         property->sequence = dbobj.sequence;
 514
 515                         reply->values.push_back(property);
 516                 }
 517         }
 518
 519         reply->success = true;
 520         reply->completed(reply);
 521
 522         delete db;
 523 }
 524
 525 AsyncPropertyReply *DatabaseSink::setProperty(AsyncSetPropertyRequest request)
 526 {
 527         AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 528         reply->success = false;
 529
 530         if(request.property == DatabaseLogging)
 531         {
 532                 if(request.value->value<bool>())
 533                 {
 534                         setPlayback(false);
 535                         startDb();
 536                         reply->success = true;
 537                         databaseLogging = true;
 538                         routingEngine->updateProperty(&databaseLogging,uuid());
 539                 }
 540                 else
 541                 {
 542                         stopDb();
 543                         reply->success = true;
 544                         databaseLogging = false;
 545                         routingEngine->updateProperty(&databaseLogging,uuid());
 546                 }
 547         }
 548
 549         else if(request.property == DatabaseFile)
 550         {
 551                 std::string fname = request.value->toString();
 552
 553                 databaseName = fname;
 554
 555                 routingEngine->updateProperty(&databaseName,uuid());
 556
 557                 reply->success = true;
 558         }
 559         else if( request.property == DatabasePlayback)
 560         {
 561                 if(request.value->value<bool>())
 562                 {
 563                         setLogging(false);
 564                         startPlayback();
 565
 566                         routingEngine->updateProperty(&playback,uuid());
 567                 }
 568                 else
 569                 {
 570                         if(playbackShared)
 571                                 playbackShared->stop = true;
 572
 573                         playback = false;
 574
 575                         routingEngine->updateProperty(&playback,uuid());
 576                 }
 577
 578                 reply->success = true;
 579         }
 580
 581         return reply;
 582 }
 583
 584 void DatabaseSink::subscribeToPropertyChanges(VehicleProperty::Property )
 585 {
 586
 587 }
 588
 589 void DatabaseSink::unsubscribeToPropertyChanges(VehicleProperty::Property )
 590 {
 591 }