plugins/common/jsonprotocol.cpp

   1 #include "jsonprotocol.h"
   2
   3 #include <jsonhelper.h>
   4 #include <listplusplus.h>
   5
   6 #include <glib.h>
   7
   8 bool readCallback(GIOChannel *source, GIOCondition condition, gpointer data)
   9 {
  10         if(condition & G_IO_ERR)
  11         {
  12                 DebugOut(DebugOut::Error)<<"BaseJsonReader polling error."<<endl;
  13         }
  14
  15         if (condition & G_IO_HUP)
  16         {
  17                 //Hang up. Returning false closes out the GIOChannel.
  18                 DebugOut(DebugOut::Warning)<<"socket hangup event..."<<endl;
  19                 return false;
  20         }
  21
  22         amb::BaseJsonMessageReader * p = static_cast<amb::BaseJsonMessageReader*>(data);
  23
  24         p->canHasData();
  25
  26         return true;
  27 }
  28
  29 amb::AmbRemoteClient::AmbRemoteClient(AbstractIo *io)
  30         :BaseJsonMessageReader(io), serverTimeOffset(0)
  31 {
  32         TimeSyncMessage timeSyncRequest;
  33
  34         send(timeSyncRequest);
  35 }
  36
  37 void amb::AmbRemoteClient::list(amb::ListCallback cb)
  38 {
  39         ListMethodCall::Ptr methodCall = ListMethodCall::create();
  40         methodCall->replyCallback = cb;
  41         mListCalls.push_back(methodCall);
  42
  43         send(methodCall);
  44 }
  45
  46 void amb::AmbRemoteClient::get(const string &objectName, amb::ObjectCallback cb)
  47 {
  48         get(objectName, "", Zone::None, cb);
  49 }
  50
  51 void amb::AmbRemoteClient::get(const string &objectName, const string &sourceUuid, amb::ObjectCallback cb)
  52 {
  53         get(objectName, sourceUuid, Zone::None, cb);
  54 }
  55
  56 void amb::AmbRemoteClient::get(const string &objectName, Zone::Type zone, amb::ObjectCallback cb)
  57 {
  58         get(objectName, "", zone, cb);
  59 }
  60
  61 void amb::AmbRemoteClient::get(const string &objectName, const string &sourceUuid, Zone::Type zone, amb::ObjectCallback cb)
  62 {
  63         GetMethodCall::Ptr getCall = GetMethodCall::create();
  64         getCall->sourceUuid = sourceUuid;
  65         getCall->zone = zone;
  66         getCall->value = amb::make_shared(new Object(objectName));
  67         getCall->replyCallback = cb;
  68
  69         mGetMethodCalls.push_back(getCall);
  70
  71         send(getCall);
  72 }
  73
  74 void amb::AmbRemoteClient::set(const string &objectName, Object::Ptr value, SetCallback cb)
  75 {
  76         set(objectName, value, "", Zone::None, cb);
  77 }
  78
  79 void amb::AmbRemoteClient::set(const string &objectName, Object::Ptr value, const string &sourceUuid, Zone::Type zone, SetCallback cb)
  80 {
  81         SetMethodCall::Ptr setCall = SetMethodCall::create();
  82         setCall->sourceUuid = sourceUuid;
  83         setCall->zone = zone;
  84         setCall->value = value;
  85         setCall->replyCallback = cb;
  86
  87         mSetMethodCalls.push_back(setCall);
  88
  89         send(setCall);
  90 }
  91
  92 const string amb::AmbRemoteClient::subscribe(const string &objectName, const string &sourceUuid, Zone::Type zone, amb::ObjectCallback cb)
  93 {
  94         std::string subscription = createSubscriptionId(objectName, sourceUuid, zone);
  95
  96         SubscribeMethodCall call(objectName);
  97
  98         Subscription sub(call, cb);
  99
 100         mSubscriptions[subscription].push_back(sub);
 101
 102         send(call);
 103
 104         return call.messageId;
 105 }
 106
 107 void amb::AmbRemoteClient::subscribe(const string &objectName, amb::ObjectCallback cb)
 108 {
 109         subscribe(objectName, "", Zone::None, cb);
 110 }
 111
 112 void amb::AmbRemoteClient::unsubscribe(const string &subscribeId)
 113 {
 114         for(auto i : mSubscriptions)
 115         {
 116                 auto subscriptions = &i.second;
 117                 for(auto n : *subscriptions)
 118                 {
 119                         if(n.subscriptionId() == subscribeId)
 120                         {
 121                                 removeOne(subscriptions, n);
 122
 123                                 if(!subscriptions->size())
 124                                 {
 125                                         UnsubscribeMethodCall call(n.call);
 126
 127                                         send(call);
 128                                 }
 129                         }
 130                 }
 131         }
 132 }
 133
 134 double amb::AmbRemoteClient::correctTimeFromServer(double serverTimestamp)
 135 {
 136         return serverTimestamp - serverTimeOffset;
 137 }
 138
 139 void amb::AmbRemoteClient::hasJsonMessage(const picojson::value &json)
 140 {
 141         DebugOut(7) << "json: " << json.serialize() << endl;
 142
 143         if(BaseMessage::is<MethodReply<MethodCall>>(json))
 144         {
 145                 if(BaseMessage::is<MethodReply<ListMethodCall>>(json))
 146                 {
 147                         MethodReply<ListMethodCall> listMethodReply;
 148                         listMethodReply.fromJson(json);
 149
 150                         const ListMethodCall::Ptr listMethod = listMethodReply.method();
 151
 152                         auto itr = std::find_if(mListCalls.begin(), mListCalls.end(),[&listMethod](auto o)
 153                         {
 154                                 return o->messageId == listMethod->messageId;
 155                         });
 156                         if(itr != mListCalls.end())
 157                         {
 158                                 auto found = *itr;
 159                                 auto cb = found->replyCallback;
 160
 161                                 try
 162                                 {
 163                                         cb(listMethod->objectNames);
 164                                 }
 165                                 catch(...)
 166                                 {
 167                                         DebugOut(DebugOut::Warning) << "callback for 'list' is not valid" << endl;
 168                                 }
 169
 170                                 mListCalls.erase(itr);
 171                         }
 172                 }
 173                 else if(BaseMessage::is<MethodReply<GetMethodCall>>(json))
 174                 {
 175                         MethodReply<GetMethodCall> reply;
 176                         reply.fromJson(json);
 177                         GetMethodCall::Ptr getCall = reply.method();
 178
 179                         auto itr = std::find_if(mGetMethodCalls.begin(), mGetMethodCalls.end(),[&getCall](auto o)
 180                         {
 181                                 return o->messageId == getCall->messageId;
 182                         });
 183
 184                         if(itr != mGetMethodCalls.end())
 185                         {
 186                                 auto found = *itr;
 187                                 auto cb = found->replyCallback;
 188
 189                                 try
 190                                 {
 191                                         cb(getCall->value);
 192                                 }
 193                                 catch(...)
 194                                 {
 195                                         DebugOut(DebugOut::Warning) << "Invalid Get callback " << endl;
 196                                 }
 197
 198                                 mGetMethodCalls.erase(itr);
 199                         }
 200                 }
 201                 else if(BaseMessage::is<MethodReply<SetMethodCall>>(json))
 202                 {
 203                         MethodReply<SetMethodCall> reply;
 204                         reply.fromJson(json);
 205
 206                         auto call = reply.method();
 207
 208                         auto itr = std::find_if(mSetMethodCalls.begin(), mSetMethodCalls.end(),[&call](auto o)
 209                         {
 210                                 return o->messageId == call->messageId;
 211                         });
 212
 213                         if(itr != mSetMethodCalls.end())
 214                         {
 215                                 auto found = *itr;
 216                                 auto cb = found->replyCallback;
 217
 218                                 try
 219                                 {
 220                                         cb(reply.methodSuccess);
 221                                 }
 222                                 catch(...)
 223                                 {
 224                                         DebugOut(DebugOut::Warning) << "Invalid Set callback " << endl;
 225                                 }
 226                                 mSetMethodCalls.erase(itr);
 227                         }
 228                 }
 229         }
 230         else if(BaseMessage::is<MethodReply<TimeSyncMessage>>(json))
 231         {
 232                 DebugOut(7) << "Received time sync message" << endl;
 233                 MethodReply<TimeSyncMessage> reply;
 234                 reply.fromJson(json);
 235
 236                 if(reply.methodSuccess)
 237                 {
 238                         serverTimeOffset = amb::Timestamp::instance()->epochTime() - reply.method()->serverTime;
 239                 }
 240                 else
 241                 {
 242                         DebugOut(DebugOut::Warning) << "Time Sync request failed" << endl;
 243                 }
 244         }
 245         else if(BaseMessage::is<EventMessage>(json))
 246         {
 247                 if(PropertyChangeEvent::is(json))
 248                 {
 249                         DebugOut(7) << "property changed event" << endl;
 250
 251                         PropertyChangeEvent::Ptr obj = PropertyChangeEvent::create();
 252                         if(!obj->fromJson(json))
 253                                 return;
 254
 255                         std::string subscribeId = createSubscriptionId(obj->value->interfaceName, obj->sourceUuid, obj->zone);
 256
 257                         if(!amb::containsKey(mSubscriptions, subscribeId))
 258                         {
 259                                 DebugOut(DebugOut::Warning) << "We haven't subscribed to this interface at this zone from this source..." << endl;
 260                                 return;
 261                         }
 262
 263                         auto list = mSubscriptions[subscribeId];
 264
 265                         for(auto i : list)
 266                         {
 267                                 i.callback(obj->value);
 268                         }
 269                 }
 270         }
 271         else
 272         {
 273                 BaseMessage msg;
 274                 msg.fromJson(json);
 275                 DebugOut(DebugOut::Warning) << "Unhandled message: " << msg.name << " type: " << msg.type << endl;
 276         }
 277 }
 278
 279 string amb::AmbRemoteClient::createSubscriptionId(const string & objectName, const string & sourceUuid, Zone::Type zone)
 280 {
 281         std::string str = std::string(objectName + sourceUuid + std::to_string(zone));
 282         return g_compute_checksum_for_string(G_CHECKSUM_MD5, str.c_str(), str.length());
 283 }
 284
 285 picojson::value amb::BaseMessage::toJson()
 286 {
 287         picojson::object val;
 288
 289         val["name"] = picojson::value(name);
 290         val["type"] = picojson::value(type);
 291         val["messageId"] = picojson::value(messageId);
 292         val["data"] = picojson::value(data);
 293
 294         return picojson::value(val);
 295 }
 296
 297 bool amb::BaseMessage::fromJson(const picojson::value &json)
 298 {
 299         if(!json.is<picojson::object>() || !json.contains("type") || !json.contains("name") || !json.contains("messageId"))
 300         {
 301                 DebugOut(DebugOut::Error) << "malformed message: is not json object or does not contain keys 'type', 'name' or 'messageId'." << endl;
 302                 return false;
 303         }
 304
 305         picojson::object obj = json.get<picojson::object>();
 306
 307         type = obj["type"].to_str();
 308         name = obj["name"].to_str();
 309         messageId = obj["messageId"].to_str();
 310
 311         if(json.contains("data"))
 312         {
 313                 data = json.get("data");
 314         }
 315
 316         return true;
 317 }
 318
 319
 320 picojson::value amb::ListMethodCall::toJson()
 321 {
 322         picojson::object v = MethodCall::toJson().get<picojson::object>();
 323
 324         picojson::array list;
 325
 326         for(auto i : objectNames)
 327         {
 328                 list.push_back(Object::toJson(i));
 329         }
 330
 331         v["data"] = picojson::value(list);
 332
 333         return picojson::value(v);
 334 }
 335
 336 bool amb::ListMethodCall::fromJson(const picojson::value &json)
 337 {
 338         if(!MethodCall::fromJson(json) || name != "list" || !data.is<picojson::array>())
 339         {
 340                 DebugOut(DebugOut::Error) << "type not 'list' or data not type json array" << endl;
 341                 return false;
 342         }
 343
 344         objectNames.clear();
 345
 346         picojson::array dataList = json.get("data").get<picojson::array>();
 347
 348         for(auto i : dataList)
 349         {
 350                 if(!i.is<picojson::object>())
 351                 {
 352                         DebugOut(DebugOut::Warning) << "Malformed data.  Expected 'object'.  Got '" << i.to_str() << "'" << endl;
 353                         continue;
 354                 }
 355                 picojson::object obj = i.get<picojson::object>();
 356
 357                 Object::Ptr ambObj = Object::fromJson(obj);
 358
 359                 objectNames.push_back(ambObj);
 360         }
 361
 362         return true;
 363 }
 364
 365
 366 amb::BaseJsonMessageReader::BaseJsonMessageReader(AbstractIo *io)
 367         :mIo(io)
 368 {
 369         GIOChannel *chan = g_io_channel_unix_new(mIo->fileDescriptor());
 370         g_io_add_watch(chan, GIOCondition(G_IO_IN | G_IO_HUP | G_IO_ERR),(GIOFunc)readCallback, this);
 371         g_io_channel_set_close_on_unref(chan, true);
 372         g_io_channel_unref(chan);
 373 }
 374
 375 void amb::BaseJsonMessageReader::canHasData()
 376 {
 377         std::string d = mIo->read();
 378         incompleteMessage += d;
 379
 380         while(hasJson());
 381 }
 382
 383 void amb::BaseJsonMessageReader::closed()
 384 {
 385         mIo->close();
 386
 387         if(disconnected)
 388                 disconnected();
 389 }
 390
 391 bool amb::BaseJsonMessageReader::hasJson()
 392 {
 393         std::string::size_type start = incompleteMessage.find("{");
 394
 395         if(start == std::string::npos && incompleteMessage.empty())
 396         {
 397                 return false;
 398         }
 399
 400         if(start > 0)
 401         {
 402                 DebugOut(7) << "We have an incomplete message at the beginning.  Toss it away:" << endl;
 403                 DebugOut(7) << incompleteMessage << endl;
 404                 incompleteMessage = incompleteMessage.substr(start-1);
 405         }
 406
 407         int end = incompleteMessage.find("\n");
 408
 409         if(end == std::string::npos)
 410         {
 411                 return false;
 412         }
 413
 414         std::string tryMessage = incompleteMessage.substr(0, end+1);
 415
 416         DebugOut(6) << "Trying to parse message: " << tryMessage << endl;
 417
 418         picojson::value doc;
 419
 420         picojson::parse(doc, tryMessage);
 421
 422         std::string parseError = picojson::get_last_error();
 423
 424         if(!parseError.empty())
 425         {
 426                 DebugOut(7) << "Invalid or incomplete message" << endl;
 427                 DebugOut(7) << parseError << endl;
 428                 return false;
 429         }
 430
 431         incompleteMessage = end == incompleteMessage.length()-1 ? "" : incompleteMessage.substr(end+1);
 432
 433         hasJsonMessage(doc);
 434         return true;
 435 }
 436
 437 picojson::value amb::MethodCall::toJson()
 438 {
 439         picojson::value value = BaseMessage::toJson();
 440
 441         picojson::object obj = value.get<picojson::object>();
 442
 443         obj["source"] = picojson::value(sourceUuid);
 444         obj["zone"] = picojson::value((double)zone);
 445
 446         return picojson::value(obj);
 447 }
 448
 449 bool amb::MethodCall::fromJson(const picojson::value &json)
 450 {
 451         if(!BaseMessage::fromJson(json))
 452                 return false;
 453
 454         sourceUuid = json.get("source").to_str();
 455         zone = json.get("zone").get<double>();
 456
 457         return true;
 458 }
 459
 460 amb::AmbRemoteServer::AmbRemoteServer(AbstractIo *io, AbstractRoutingEngine *re)
 461         :BaseJsonMessageReader(io), routingEngine(re)
 462 {
 463
 464 }
 465
 466 void amb::AmbRemoteServer::list(ListMethodCall::Ptr call)
 467 {
 468
 469 }
 470
 471 void amb::AmbRemoteServer::get(GetMethodCall::Ptr get)
 472 {
 473
 474 }
 475
 476 void amb::AmbRemoteServer::set(SetMethodCall::Ptr set)
 477 {
 478
 479 }
 480
 481 void amb::AmbRemoteServer::subscribe(SubscribeMethodCall::Ptr call)
 482 {
 483
 484 }
 485
 486 void amb::AmbRemoteServer::unsubscribe(amb::UnsubscribeMethodCall::Ptr call)
 487 {
 488
 489 }
 490
 491 void amb::AmbRemoteServer::hasJsonMessage(const picojson::value &json)
 492 {
 493         DebugOut(7) << "json: " << json.serialize() << endl;
 494
 495         if(!BaseMessage::validate(json))
 496         {
 497                 DebugOut(DebugOut::Warning) << "not a valid message: " << json.serialize() << endl;
 498                 return;
 499         }
 500
 501         if(BaseMessage::is<MethodCall>(json))
 502         {
 503                 if(BaseMessage::is<ListMethodCall>(json))
 504                 {
 505                         ListMethodCall::Ptr listCall = ListMethodCall::create();
 506                         listCall->fromJson(json);
 507
 508                         list(listCall);
 509                 }
 510                 else if(BaseMessage::is<GetMethodCall>(json))
 511                 {
 512                         GetMethodCall::Ptr getCall = GetMethodCall::create();
 513                         getCall->fromJson(json);
 514
 515                         get(getCall);
 516                 }
 517                 else if(BaseMessage::is<SetMethodCall>(json))
 518                 {
 519                         SetMethodCall::Ptr setCall = SetMethodCall::create();
 520                         setCall->fromJson(json);
 521
 522                         set(setCall);
 523                 }
 524                 else if(BaseMessage::is<SubscribeMethodCall>(json))
 525                 {
 526                         SubscribeMethodCall::Ptr call = SubscribeMethodCall::create();
 527                         call->fromJson(json);
 528
 529                         subscribe(call);
 530                 }
 531                 else if(BaseMessage::is<UnsubscribeMethodCall>(json))
 532                 {
 533                         UnsubscribeMethodCall::Ptr call = UnsubscribeMethodCall::create();
 534                         call->fromJson(json);
 535
 536                         unsubscribe(call);
 537                 }
 538                 else
 539                 {
 540                         BaseMessage call;
 541                         call.fromJson(json);
 542                         DebugOut(DebugOut::Warning) << "Unhandled method call: " << call.name << endl;
 543                 }
 544         }
 545         else if(BaseMessage::is<TimeSyncMessage>(json))
 546         {
 547                 TimeSyncMessage::Ptr call = TimeSyncMessage::create();
 548                 call->fromJson(json);
 549
 550                 call->serverTime = amb::Timestamp::instance()->epochTime();
 551
 552                 MethodReply<TimeSyncMessage> reply(call, true);
 553
 554                 send(reply);
 555         }
 556         else
 557         {
 558                 BaseMessage message;
 559                 message.fromJson(json);
 560
 561                 DebugOut(DebugOut::Warning) << "Unhandled message: type: " << message.type << " name: " << message.name << endl;
 562         }
 563 }
 564
 565 picojson::value amb::GetMethodCall::toJson()
 566 {
 567         picojson::value val = MethodCall::toJson();
 568
 569         picojson::object obj = val.get<picojson::object>();
 570
 571         obj["data"] = Object::toJson(value);
 572
 573         return picojson::value(obj);
 574 }
 575
 576 bool amb::GetMethodCall::fromJson(const picojson::value &json)
 577 {
 578         MethodCall::fromJson(json);
 579
 580         value = Object::fromJson(json.get("data").get<picojson::object>());
 581 }
 582
 583
 584 amb::Object::Ptr amb::Object::fromJson(const picojson::object &obj)
 585 {
 586         if(!amb::containsKey(obj, "interfaceName"))
 587         {
 588                 DebugOut(DebugOut::Warning) << "object missing interfaceName" << endl;
 589                 return Object::Ptr(new Object());
 590         }
 591         Object * ambObj = new Object(obj.at("interfaceName").to_str());
 592
 593         for(auto i : obj)
 594         {
 595                 if(i.second.is<picojson::object>())
 596                 {
 597                         (*ambObj)[i.first] = std::shared_ptr<AbstractPropertyType>(amb::jsonToProperty(i.second));
 598                 }
 599         }
 600
 601         return Object::Ptr(ambObj);
 602 }
 603
 604 picojson::value amb::Object::toJson(const Object::Ptr &obj)
 605 {
 606         picojson::object jsonObj;
 607         jsonObj["interfaceName"] = picojson::value(obj->interfaceName);
 608         for(auto i : *obj.get())
 609         {
 610                 jsonObj[i.first] = i.second->toJson();
 611         }
 612
 613         return picojson::value(jsonObj);
 614 }
 615
 616
 617 picojson::value amb::SetMethodCall::toJson()
 618 {
 619         picojson::value val = MethodCall::toJson();
 620
 621         picojson::object obj = val.get<picojson::object>();
 622
 623         obj["data"] = Object::toJson(value);
 624
 625         return picojson::value(obj);
 626 }
 627
 628 bool amb::SetMethodCall::fromJson(const picojson::value &json)
 629 {
 630         MethodCall::fromJson(json);
 631
 632         value = Object::fromJson(json.get("data").get<picojson::object>());
 633
 634         return true;
 635 }
 636
 637
 638 picojson::value amb::SubscribeMethodCall::toJson()
 639 {
 640         auto json = MethodCall::toJson();
 641
 642         auto obj = json.get<picojson::object>();
 643
 644         obj["interfaceName"] = picojson::value(interfaceName);
 645
 646         return picojson::value(obj);
 647 }
 648
 649 bool amb::SubscribeMethodCall::fromJson(const picojson::value &json)
 650 {
 651         if(!MethodCall::fromJson(json))
 652                 return false;
 653
 654         interfaceName = json.get("interfaceName").to_str();
 655
 656         return true;
 657 }
 658
 659 picojson::value amb::UnsubscribeMethodCall::toJson()
 660 {
 661         auto json = MethodCall::toJson();
 662
 663         auto obj = json.get<picojson::object>();
 664
 665         obj["interfaceName"] = picojson::value(interfaceName);
 666
 667         return picojson::value(obj);
 668 }
 669
 670 bool amb::UnsubscribeMethodCall::fromJson(const picojson::value &json)
 671 {
 672         if(!MethodCall::fromJson(json))
 673                 return false;
 674
 675         interfaceName = json.get("interfaceName").to_str();
 676
 677         return true;
 678 }
 679
 680
 681 picojson::value amb::TimeSyncMessage::toJson()
 682 {
 683         auto val = BaseMessage::toJson();
 684
 685         auto obj = val.get<picojson::object>();
 686
 687         obj["serverTime"] = picojson::value(serverTime);
 688
 689         return picojson::value(obj);
 690 }
 691
 692 bool amb::TimeSyncMessage::fromJson(const picojson::value &json)
 693 {
 694         if(!BaseMessage::fromJson(json))
 695                 return false;
 696
 697         serverTime = json.get("serverTime").get<double>();
 698
 699         return true;
 700 }
 701
 702
 703 picojson::value amb::PropertyChangeEvent::toJson()
 704 {
 705         auto val = EventMessage::toJson();
 706
 707         auto obj = val.get<picojson::object>();
 708         obj["data"] = Object::toJson(value);
 709         obj["zone"] = picojson::value((double)zone);
 710         obj["source"] = picojson::value(sourceUuid);
 711
 712         return picojson::value(obj);
 713 }
 714
 715 bool amb::PropertyChangeEvent::fromJson(const picojson::value &json)
 716 {
 717         if(!EventMessage::fromJson(json))
 718                 return false;
 719
 720         value = Object::fromJson(json.get("data").get<picojson::object>());
 721
 722         return true;
 723 }