plugins/database/databasesink.cpp

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