plugins/websocket/websocketsource.cpp

   1 /*
   2         Copyright (C) 2012  Intel Corporation
   3
   4         This library is free software; you can redistribute it and/or
   5         modify it under the terms of the GNU Lesser General Public
   6         License as published by the Free Software Foundation; either
   7         version 2.1 of the License, or (at your option) any later version.
   8
   9         This library is distributed in the hope that it will be useful,
  10         but WITHOUT ANY WARRANTY; without even the implied warranty of
  11         MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  12         Lesser General Public License for more details.
  13
  14         You should have received a copy of the GNU Lesser General Public
  15         License along with this library; if not, write to the Free Software
  16         Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301  USA
  17 */
  18
  19
  20 #include "websocketsource.h"
  21 #include <iostream>
  22 #include <boost/assert.hpp>
  23 #include <boost/lexical_cast.hpp>
  24 #include <glib.h>
  25 #include <sstream>
  26 #include <listplusplus.h>
  27 #include <memory>
  28 #include <timestamp.h>
  29 #include "uuidhelper.h"
  30
  31 #include <QVariantMap>
  32 #include <QJsonDocument>
  33 #include <QStringList>
  34
  35 #include "debugout.h"
  36 #include "common.h"
  37 #include "superptr.hpp"
  38
  39 #define __SMALLFILE__ std::string(__FILE__).substr(std::string(__FILE__).rfind("/")+1)
  40 lws_context *context = NULL;
  41 WebSocketSource *source;
  42 AbstractRoutingEngine *m_re;
  43
  44 double oldTimestamp=0;
  45 double totalTime=0;
  46 double numUpdates=0;
  47 double averageLatency=0;
  48
  49 class UniquePropertyCache
  50 {
  51 public:
  52         bool hasProperty(std::string name, std::string source, Zone::Type zone)
  53         {
  54                 for(auto i : mProperties)
  55                 {
  56                         if(i->name == name &&
  57                                         i->sourceUuid == source &&
  58                                         i->zone == zone)
  59                         {
  60                                 return true;
  61                         }
  62                 }
  63                 return false;
  64         }
  65
  66         std::shared_ptr<AbstractPropertyType> append(std::string name, std::string source, Zone::Type zone, std::string type)
  67         {
  68                 for(auto i : mProperties)
  69                 {
  70                         if(i->name == name &&
  71                                         i->sourceUuid == source &&
  72                                         i->zone == zone)
  73                         {
  74                                 return i;
  75                         }
  76                 }
  77
  78                 auto t = VehicleProperty::getPropertyTypeForPropertyNameValue(name);
  79
  80                 if(!t)
  81                 {
  82                         VehicleProperty::registerProperty(name, [name, type]() -> AbstractPropertyType* {
  83                                 if(type == amb::BasicTypes::UInt16Str)
  84                                 {
  85                                         return new BasicPropertyType<uint16_t>(name, 0);
  86                                 }
  87                                 else if(type == amb::BasicTypes::Int16Str)
  88                                 {
  89                                         return new BasicPropertyType<int16_t>(name, 0);
  90                                 }
  91                                 else if(type == amb::BasicTypes::UInt32Str)
  92                                 {
  93                                         return new BasicPropertyType<uint32_t>(name, 0);
  94                                 }
  95                                 else if(type == amb::BasicTypes::Int32Str)
  96                                 {
  97                                         return new BasicPropertyType<int32_t>(name, 0);
  98                                 }
  99                                 else if(type == amb::BasicTypes::StringStr)
 100                                 {
 101                                         return new StringPropertyType(name);
 102                                 }
 103                                 else if(type == amb::BasicTypes::DoubleStr)
 104                                 {
 105                                         return new BasicPropertyType<double>(name, 0);
 106                                 }
 107                                 else if(type == amb::BasicTypes::BooleanStr)
 108                                 {
 109                                         return new BasicPropertyType<bool>(name, false);
 110                                 }
 111                                 DebugOut(DebugOut::Warning) << "Unknown or unsupported type: " << type << endl;
 112                                 return nullptr;
 113                         });
 114                         t = VehicleProperty::getPropertyTypeForPropertyNameValue(name);
 115                 }
 116
 117                 if(t)/// check again to see if registration succeeded
 118                 {
 119                         t->sourceUuid = source;
 120                         t->zone = zone;
 121
 122                         mProperties.emplace_back(t);
 123                 }
 124
 125                 return property(name, source, zone); /// will return nullptr if t didn't register
 126         }
 127
 128         std::shared_ptr<AbstractPropertyType> property(std::string name, std::string source, Zone::Type zone)
 129         {
 130                 for(auto i : mProperties)
 131                 {
 132                         if(i->name == name &&
 133                                         i->sourceUuid == source &&
 134                                         i->zone == zone)
 135                         {
 136                                 return i;
 137                         }
 138                 }
 139
 140                 return nullptr;
 141         }
 142
 143         std::vector<std::shared_ptr<AbstractPropertyType>> properties() { return mProperties; }
 144
 145 private:
 146         std::vector<std::shared_ptr<AbstractPropertyType>> mProperties;
 147 };
 148
 149 UniquePropertyCache properties;
 150
 151 static int callback_http_only(struct lws *wsi,enum lws_callback_reasons reason,void *user, void *in, size_t len);
 152 static struct lws_protocols protocols[] = {
 153         {
 154                 "http-only",
 155                 callback_http_only,
 156                 0,
 157                 128,
 158                 0,
 159                 NULL,
 160         },
 161         {
 162                 NULL,
 163                 NULL,
 164                 0,
 165                 0,
 166                 0,
 167                 NULL,
 168         }
 169 };
 170
 171 //Called when a client connects, subscribes, or unsubscribes.
 172 void WebSocketSource::checkSubscriptions()
 173 {
 174         while (queuedRequests.size() > 0)
 175         {
 176                 VehicleProperty::Property prop = queuedRequests.front();
 177                 removeOne(&queuedRequests,prop);
 178                 if (contains(activeRequests,prop))
 179                 {
 180                         return;
 181                 }
 182                 activeRequests.push_back(prop);
 183
 184                 QVariantMap reply;
 185
 186                 reply["type"] = "method";
 187                 reply["name"] = "subscribe";
 188                 reply["property"] = prop.c_str();
 189                 reply["transactionid"] = "d293f670-f0b3-11e1-aff1-0800200c9a66";
 190
 191                 lwsWriteVariant(clientsocket, reply);
 192         }
 193 }
 194 void WebSocketSource::setConfiguration(map<string, string> config)
 195 {
 196         //printf("WebSocketSource::setConfiguration has been called\n");
 197         std::string ip;
 198         int port;
 199         configuration = config;
 200
 201         if(config.find("binaryProtocol") != config.end())
 202         {
 203                 doBinary = config["binaryProtocol"] == "true";
 204         }
 205
 206         for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
 207         {
 208                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting for WebSocketSource:" << (*i).first << ":" << (*i).second << "\n";
 209                 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
 210                 if ((*i).first == "ip")
 211                 {
 212                         ip = (*i).second;
 213                 }
 214                 if ((*i).first == "port")
 215                 {
 216                         port = boost::lexical_cast<int>((*i).second);
 217                 }
 218                 if ((*i).first == "ssl")
 219                 {
 220                         if ((*i).second == "true")
 221                         {
 222                                 m_sslEnabled = true;
 223                         }
 224                         else
 225                         {
 226                                 m_sslEnabled = false;
 227                         }
 228                 }
 229         }
 230         //printf("Connecting to websocket server at %s port %i\n",ip.c_str(),port);
 231         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Connecting to websocket server at" << ip << ":" << port << "\n";
 232         int sslval = 0;
 233         if (m_sslEnabled)
 234         {
 235                 DebugOut(5) << "SSL ENABLED" << endl;
 236                 sslval = 2;
 237         }
 238
 239         clientsocket = lws_client_connect(context, ip.c_str(), port, sslval,"/", "localhost", "websocket", protocols[0].name, -1);
 240 }
 241
 242 PropertyInfo WebSocketSource::getPropertyInfo(const VehicleProperty::Property &property)
 243 {
 244         Zone::ZoneList zones;
 245         for(auto i : properties.properties())
 246         {
 247                 if(i->name == property)
 248                 {
 249                         zones.push_back(i->zone);
 250                 }
 251         }
 252
 253         return PropertyInfo(0, zones);
 254 }
 255
 256 bool gioPollingFunc(GIOChannel *source, GIOCondition condition, gpointer data)
 257 {
 258         //This is the polling function. If it return false, glib will stop polling this FD.
 259
 260         oldTimestamp = amb::currentTime();
 261
 262         struct pollfd pollstruct;
 263         int newfd = g_io_channel_unix_get_fd(source);
 264         pollstruct.fd = newfd;
 265         pollstruct.events = condition;
 266         pollstruct.revents = condition;
 267         lws_service_fd(context,&pollstruct);
 268         if (condition & G_IO_HUP)
 269         {
 270                 //Hang up. Returning false closes out the GIOChannel.
 271                 //printf("Callback on G_IO_HUP\n");
 272                 return false;
 273         }
 274         if (condition & G_IO_IN)
 275         {
 276
 277         }
 278         DebugOut() << "gioPollingFunc" << condition << endl;
 279
 280         return true;
 281 }
 282
 283 static int checkTimeouts(gpointer data)
 284 {
 285         WebSocketSource *src = (WebSocketSource*)data;
 286         for (auto i=src->uuidTimeoutMap.begin();i!= src->uuidTimeoutMap.end();i++)
 287         {
 288                 if (src->uuidRangedReplyMap.find((*i).first) != src->uuidRangedReplyMap.end())
 289                 {
 290                         //A source exists!
 291                         if (amb::currentTime() > (*i).second)
 292                         {
 293                                 //We've reached timeout
 294                                 DebugOut() << "Timeout reached for request ID:" << (*i).first << "\n";
 295                                 src->uuidRangedReplyMap[(*i).first]->success = false;
 296                                 src->uuidRangedReplyMap[(*i).first]->completed(src->uuidRangedReplyMap[(*i).first]);
 297                                 src->uuidRangedReplyMap.erase((*i).first);
 298                                 src->uuidTimeoutMap.erase((*i).first);
 299                                 i--;
 300
 301                                 if (src->uuidTimeoutMap.size() == 0)
 302                                 {
 303                                         return 0;
 304                                 }
 305
 306                         }
 307                         else
 308                         {
 309                                 //No timeout yet, keep waiting.
 310                         }
 311                 }
 312                 else
 313                 {
 314                         //Reply has already come back, ignore and erase from list.
 315                         src->uuidTimeoutMap.erase((*i).first);
 316                         i--;
 317
 318                         if (src->uuidTimeoutMap.size() == 0)
 319                         {
 320                                 return 0;
 321                         }
 322                 }
 323
 324         }
 325         return 0;
 326 }
 327
 328 static int callback_http_only(struct lws *wsi,enum lws_callback_reasons reason, void *user, void *in, size_t len)
 329 {
 330         unsigned char buf[LWS_SEND_BUFFER_PRE_PADDING + 4096 + LWS_SEND_BUFFER_POST_PADDING];
 331         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << reason << "callback_http_only" << endl;
 332         switch (reason)
 333         {
 334                 case LWS_CALLBACK_CLOSED:
 335                         //fprintf(stderr, "mirror: LWS_CALLBACK_CLOSED\n");
 336                         //wsi_mirror = NULL;
 337                         //printf("Connection closed!\n");
 338                         break;
 339
 340                         //case LWS_CALLBACK_PROTOCOL_INIT:
 341                 case LWS_CALLBACK_CLIENT_ESTABLISHED:
 342                 {
 343                         //This happens when a client initally connects. We need to request the support event types.
 344                         source->clientConnected = true;
 345                         source->checkSubscriptions();
 346                         //printf("Incoming connection!\n");
 347                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming connection" << endl;
 348
 349                         QVariantMap toSend;
 350                         toSend["type"] = "method";
 351                         toSend["name"] = "getSupported";
 352                         toSend["transactionid"] = amb::createUuid().c_str();
 353
 354                         lwsWriteVariant(wsi, toSend);
 355
 356                         break;
 357                 }
 358                 case LWS_CALLBACK_CLIENT_RECEIVE:
 359                 {
 360                         QByteArray d((char*)in, len);
 361
 362                         WebSocketSource * manager = source;
 363
 364                         if(manager->expectedMessageFrames && manager->partialMessageIndex < manager->expectedMessageFrames)
 365                         {
 366                                 manager->incompleteMessage += d;
 367                                 manager->partialMessageIndex++;
 368                                 break;
 369                         }
 370                         else if(manager->expectedMessageFrames && manager->partialMessageIndex == manager->expectedMessageFrames)
 371                         {
 372                                 d = manager->incompleteMessage + d;
 373                                 manager->expectedMessageFrames = 0;
 374                         }
 375
 376                         DebugOut(7) << "data received: " << d.data() << endl;
 377
 378                         int start = d.indexOf("{");
 379
 380                         if(manager->incompleteMessage.isEmpty() && start > 0)
 381                         {
 382                                 DebugOut(7)<< "We have an incomplete message at the beginning.  Toss it away." << endl;
 383                                 d = d.right(start-1);
 384                         }
 385
 386
 387                         int end = d.lastIndexOf("}");
 388
 389                         if(end == -1)
 390                         {
 391                                 manager->incompleteMessage += d;
 392                                 break;
 393                         }
 394
 395                         QByteArray tryMessage = manager->incompleteMessage + d.left(end+1);
 396
 397                         DebugOut(6) << "Trying to parse message: " << tryMessage.data() << endl;
 398
 399                         QJsonDocument doc;
 400
 401                         QJsonParseError parseError;
 402
 403                         doc = QJsonDocument::fromJson(tryMessage, &parseError);
 404
 405                         if(doc.isNull())
 406                         {
 407                                 DebugOut(7) << "Invalid or incomplete message" << endl;
 408                                 DebugOut(7) << parseError.errorString().toStdString() << ": " << parseError.offset << endl;
 409                                 manager->incompleteMessage += d;
 410                                 break;
 411                         }
 412
 413                         manager->incompleteMessage = end == d.length()-1 ? "" : d.right(end);
 414
 415                         QVariantMap call = doc.toVariant().toMap();
 416
 417                         string type = call["type"].toString().toStdString();
 418                         string name = call["name"].toString().toStdString();
 419                         string id = call["transactionid"].toString().toStdString();
 420
 421                         if(type == "multiframe")
 422                         {
 423                                 manager->expectedMessageFrames = call["frames"].toInt();
 424                                 manager->partialMessageIndex = 1;
 425                                 manager->incompleteMessage = "";
 426
 427                         }
 428                         else if (type == "valuechanged")
 429                         {
 430                                 QVariantMap data = call["data"].toMap();
 431
 432                                 string value = data["value"].toString().toStdString();
 433                                 double timestamp = data["timestamp"].toDouble();
 434                                 int sequence = data["sequence"].toInt();
 435                                 Zone::Type zone = data["zone"].toInt();
 436                                 string type = data["type"].toString().toStdString();
 437
 438                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Value changed:" << name << value << endl;
 439
 440                                 try
 441                                 {
 442                                         auto property = properties.append(name, source->uuid(), zone, type);
 443
 444                                         if(!property)
 445                                         {
 446                                                 DebugOut(DebugOut::Warning) << "We either don't have this or don't support it ("
 447                                                                                                         << name << "," << zone << "," << type << ")" << endl;
 448                                         }
 449
 450                                         property->timestamp = timestamp;
 451                                         property->sequence = sequence;
 452                                         property->fromString(value);
 453
 454                                         m_re->updateProperty(property.get(), source->uuid());
 455
 456                                         double currenttime = amb::currentTime();
 457
 458                                         /** This is now the latency between when something is available to read on the socket, until
 459                                          *  a property is about to be updated in AMB.  This includes libwebsockets parsing and the
 460                                          *  JSON parsing in this section.
 461                                          */
 462
 463                                         DebugOut()<<"websocket network + parse latency: "<<(currenttime - property->timestamp)*1000<<"ms"<<endl;
 464                                         totalTime += (currenttime - oldTimestamp)*1000;
 465                                         numUpdates ++;
 466                                         averageLatency = totalTime / numUpdates;
 467
 468                                         DebugOut()<<"Average parse latency: "<<averageLatency<<endl;
 469                                 }
 470                                 catch (exception ex)
 471                                 {
 472                                         //printf("Exception %s\n",ex.what());
 473                                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Exception:" << ex.what() << "\n";
 474                                 }
 475                         }
 476                         else if (type == "methodReply")
 477                         {
 478                                 if (name == "getSupported" || name == "supportedChanged")
 479                                 {
 480
 481                                         QVariant data = call["data"];
 482
 483                                         QVariantList supported = data.toList();
 484
 485                                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Got getSupported request"<<endl;
 486
 487                                         double serverTime = call["systemTime"].toDouble();
 488
 489                                         DebugOut() << "Server time is: " << serverTime << endl;
 490
 491                                         if(serverTime)
 492                                                 source->serverTimeOffset = amb::Timestamp::instance()->epochTime() - serverTime;
 493
 494                                         Q_FOREACH(QVariant p, supported)
 495                                         {
 496                                                 QVariantMap d = p.toMap();
 497                                                 Zone::Type zone = d["zone"].toInt();
 498                                                 std::string name = d["property"].toString().toStdString();
 499                                                 std::string proptype = d["type"].toString().toStdString();
 500                                                 std::string source = d["source"].toString().toStdString();
 501
 502                                                 properties.append(name, source, zone, proptype);
 503                                         }
 504
 505                                         source->updateSupported();
 506
 507                                 }
 508                                 else if (name == "getRanged")
 509                                 {
 510                                         QVariantList data = call["data"].toList();
 511
 512                                         std::list<AbstractPropertyType*> propertylist;
 513
 514                                         Q_FOREACH(QVariant d, data)
 515                                         {
 516                                                 QVariantMap obj = d.toMap();
 517
 518                                                 std::string name = obj["property"].toString().toStdString();
 519                                                 std::string value = obj["value"].toString().toStdString();
 520                                                 double timestamp = obj["timestamp"].toDouble() + source->serverTimeOffset;
 521                                                 int sequence = obj["sequence"].toInt();
 522
 523                                                 AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(name, value);
 524                                                 if(!type)
 525                                                 {
 526                                                         DebugOut() << "TODO: support custom types here: " << endl;
 527                                                         continue;
 528                                                 }
 529                                                 type->timestamp = timestamp;
 530                                                 type->sequence = sequence;
 531
 532                                                 propertylist.push_back(type);
 533                                         }
 534
 535                                         if (source->uuidRangedReplyMap.find(id) != source->uuidRangedReplyMap.end())
 536                                         {
 537                                                 source->uuidRangedReplyMap[id]->values = propertylist;
 538                                                 source->uuidRangedReplyMap[id]->success = true;
 539                                                 source->uuidRangedReplyMap[id]->completed(source->uuidRangedReplyMap[id]);
 540                                                 source->uuidRangedReplyMap.erase(id);
 541                                         }
 542                                         else
 543                                         {
 544                                                 DebugOut() << "getRanged methodReply has been recieved, without a request being in!. This is likely due to a request coming in after the timeout has elapsed.\n";
 545                                         }
 546                                 }
 547                                 else if (name == "get")
 548                                 {
 549
 550                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "Got \"GET\" reply" << endl;
 551                                         if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end())
 552                                         {
 553                                                 QVariantMap obj = call["data"].toMap();
 554
 555                                                 std::string property = obj["property"].toString().toStdString();
 556                                                 std::string value = obj["value"].toString().toStdString();
 557                                                 double timestamp = obj["timestamp"].toDouble();
 558                                                 int sequence = obj["sequence"].toInt();
 559                                                 Zone::Type zone = obj["zone"].toInt();
 560
 561                                                 auto v = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(property, value));
 562
 563                                                 v->timestamp = timestamp;
 564                                                 v->sequence = sequence;
 565                                                 v->zone = zone;
 566
 567                                                 if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end() && source->uuidReplyMap[id]->error != AsyncPropertyReply::Timeout)
 568                                                 {
 569                                                         source->uuidReplyMap[id]->value = v.get();
 570                                                         source->uuidReplyMap[id]->success = true;
 571                                                         source->uuidReplyMap[id]->completed(source->uuidReplyMap[id]);
 572                                                         source->uuidReplyMap.erase(id);
 573
 574                                                 }
 575                                                 else
 576                                                 {
 577                                                         DebugOut() << "get methodReply has been recieved, without a request being in!. This is likely due to a request coming in after the timeout has elapsed.\n";
 578                                                 }
 579                                         }
 580                                         else
 581                                         {
 582                                                 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "GET Method Reply INVALID! Multiple properties detected, only single are supported!!!" << "\n";
 583                                         }
 584
 585
 586                                 }
 587                                 else if (name == "set")
 588                                 {
 589                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "Got \"SET\" event" << endl;
 590                                         std::string id = call["transactionid"].toString().toStdString();
 591
 592                                         if(source->setReplyMap.find(id) != source->setReplyMap.end() && source->setReplyMap[id]->error != AsyncPropertyReply::Timeout)
 593                                         {
 594                                                 AsyncPropertyReply* reply = source->setReplyMap[id];
 595
 596                                                 reply->success = call["success"].toBool();
 597                                                 reply->error = AsyncPropertyReply::strToError(call["error"].toString().toStdString());
 598
 599                                                 QVariantMap obj = call["data"].toMap();
 600
 601                                                 std::string property = obj["property"].toString().toStdString();
 602                                                 std::string value = obj["value"].toString().toStdString();
 603
 604                                                 double timestamp = obj["timestamp"].toDouble();
 605                                                 int sequence = obj["sequence"].toInt();
 606                                                 Zone::Type zone = obj["zone"].toInt();
 607
 608                                                 auto v = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(property, value));
 609
 610                                                 if(v)
 611                                                 {
 612                                                         v->timestamp = timestamp;
 613                                                         v->sequence = sequence;
 614                                                         v->zone = zone;
 615                                                 }
 616                                                 else
 617                                                 {
 618                                                         throw std::runtime_error("property may not be registered.");
 619                                                 }
 620
 621                                                 reply->value = v.get();
 622                                                 reply->completed(reply);
 623                                                 source->setReplyMap.erase(id);
 624                                         }
 625                                 }
 626                                 else
 627                                 {
 628                                         DebugOut(DebugOut::Warning) << "Unhandled methodReply: " << name << endl;
 629                                 }
 630
 631                         }
 632
 633                         break;
 634
 635                 }
 636                 case LWS_CALLBACK_CLIENT_CONFIRM_EXTENSION_SUPPORTED:
 637                 {
 638                         //printf("Requested extension: %s\n",(char*)in);
 639                         return 0;
 640                         break;
 641                 }
 642                 case LWS_CALLBACK_ADD_POLL_FD:
 643                 {
 644                         DebugOut(5) << __SMALLFILE__ << ":" << __LINE__ << "Adding poll for websocket IO channel" << endl;
 645                         //Add a FD to the poll list.
 646                         GIOChannel *chan = g_io_channel_unix_new(lws_get_socket_fd(wsi));
 647
 648                         /// TODO: I changed this to be more consistent with the websocket sink end. it may not be correct. TEST
 649
 650                         g_io_add_watch(chan,GIOCondition(G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP),(GIOFunc)gioPollingFunc,0);
 651                         g_io_channel_set_close_on_unref(chan,true);
 652                         g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
 653
 654                         break;
 655                 }
 656                         return 0;
 657         }
 658 }
 659 void WebSocketSource::updateSupported()
 660 {
 661         PropertyList list;
 662         for(auto i : properties.properties())
 663         {
 664                 if(!contains(list, i->name))
 665                         list.push_back(i->name);
 666         }
 667
 668         m_re->updateSupported(list, PropertyList(), this);
 669 }
 670
 671 WebSocketSource::WebSocketSource(AbstractRoutingEngine *re, map<string, string> config) : AbstractSource(re, config), partialMessageIndex(0),expectedMessageFrames(0),
 672         serverTimeOffset(0)
 673 {
 674         m_sslEnabled = false;
 675         clientConnected = false;
 676         source = this;
 677         m_re = re;
 678         struct lws_context_creation_info info;
 679         memset(&info, 0, sizeof info);
 680         info.protocols = protocols;
 681         info.extensions = nullptr;
 682
 683         if(config.find("useExtensions") != config.end() && config["useExtensions"] == "true")
 684         {
 685                 info.extensions = lws_get_internal_extensions();
 686         }
 687
 688         info.gid = -1;
 689         info.uid = -1;
 690         info.port = CONTEXT_PORT_NO_LISTEN;
 691         info.user = this;
 692
 693         context = lws_create_context(&info);
 694
 695         setConfiguration(config);
 696
 697         //printf("websocketsource loaded!!!\n");
 698         g_timeout_add(1000,checkTimeouts,this); //Do this once per second, check for functions that have timed out and reply with success = false;
 699
 700 }
 701 PropertyList WebSocketSource::supported()
 702 {
 703         PropertyList list;
 704         for(auto i : properties.properties())
 705         {
 706                 list.push_back(i->name);
 707         }
 708         return list;
 709 }
 710
 711 int WebSocketSource::supportedOperations()
 712 {
 713         /// TODO: need to do this correctly based on what the host supports.
 714         return Get | Set | GetRanged;
 715 }
 716
 717 const string WebSocketSource::uuid()
 718 {
 719         return "d293f670-f0b3-11e1-aff1-0800200c9a66";
 720 }
 721
 722 void WebSocketSource::subscribeToPropertyChanges(VehicleProperty::Property property)
 723 {
 724         //printf("Subscribed to property: %s\n",property.c_str());
 725         queuedRequests.push_back(property);
 726         if (clientConnected)
 727         {
 728                 checkSubscriptions();
 729         }
 730 }
 731
 732
 733 void WebSocketSource::unsubscribeToPropertyChanges(VehicleProperty::Property property)
 734 {
 735         removeRequests.push_back(property);
 736         if (clientConnected)
 737         {
 738                 checkSubscriptions();
 739         }
 740 }
 741
 742
 743 void WebSocketSource::getPropertyAsync(AsyncPropertyReply *reply)
 744 {
 745         std::string uuid = amb::createUuid();
 746         uuidReplyMap[uuid] = reply;
 747         uuidTimeoutMap[uuid] = amb::currentTime() + 10.0; ///TODO: 10 second timeout, make this configurable?
 748
 749         QVariantMap data;
 750         data["property"] = reply->property.c_str();
 751         data["zone"] = reply->zoneFilter;
 752
 753         QVariantMap replyvar;
 754         replyvar["type"] = "method";
 755         replyvar["name"] = "get";
 756         replyvar["data"] = data;
 757         replyvar["transactionid"] = uuid.c_str();
 758
 759         lwsWriteVariant(clientsocket, replyvar);
 760 }
 761
 762 void WebSocketSource::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 763 {
 764         std::string uuid = amb::createUuid();
 765         uuidRangedReplyMap[uuid] = reply;
 766         uuidTimeoutMap[uuid] = amb::currentTime() + 60; ///TODO: 60 second timeout, make this configurable?
 767         stringstream s;
 768         s.precision(15);
 769         s << "{\"type\":\"method\",\"name\":\"getRanged\",\"data\": {";
 770
 771         QVariantMap replyvar;
 772         replyvar["type"] = "method";
 773         replyvar["name"] = "getRanged";
 774         replyvar["transactionid"] = uuid.c_str();
 775         replyvar["timeBegin"] = reply->timeBegin - serverTimeOffset;
 776         replyvar["timeEnd"] = reply->timeEnd - serverTimeOffset;
 777         replyvar["sequenceBegin"] = reply->sequenceBegin;
 778         replyvar["sequenceEnd"] = reply->sequenceEnd;
 779
 780         QStringList properties;
 781
 782         for (auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
 783         {
 784                 VehicleProperty::Property p = *itr;
 785                 properties.append(p.c_str());
 786         }
 787
 788         replyvar["data"] = properties;
 789
 790         lwsWriteVariant(clientsocket, replyvar);
 791 }
 792
 793 AsyncPropertyReply * WebSocketSource::setProperty( AsyncSetPropertyRequest request )
 794 {
 795         AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 796
 797         std::string uuid = amb::createUuid();
 798
 799         QVariantMap data;
 800         data["property"] = request.property.c_str();
 801         data["value"] = request.value->toString().c_str();
 802         data["zone"] = request.zoneFilter;
 803
 804         QVariantMap replyvar;
 805         replyvar["type"] = "method";
 806         replyvar["name"] = "set";
 807         replyvar["data"] = data;
 808         replyvar["transactionid"] = uuid.c_str();
 809
 810         lwsWriteVariant(clientsocket, replyvar);
 811
 812         setReplyMap[uuid] = reply;
 813
 814         return reply;
 815 }
 816
 817 extern "C" AbstractSource * create(AbstractRoutingEngine* routingengine, map<string, string> config)
 818 {
 819         return new WebSocketSource(routingengine, config);
 820
 821 }