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 libwebsocket_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(libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason,void *user, void *in, size_t len);
 152 static struct libwebsocket_protocols protocols[] = {
 153         {
 154                 "http-only",
 155                 callback_http_only,
 156                 0,
 157                 128,
 158         },
 159         {  /* end of list */
 160                 NULL,
 161                 NULL,
 162                 0,
 163                 0
 164         }
 165 };
 166
 167 //Called when a client connects, subscribes, or unsubscribes.
 168 void WebSocketSource::checkSubscriptions()
 169 {
 170         while (queuedRequests.size() > 0)
 171         {
 172                 VehicleProperty::Property prop = queuedRequests.front();
 173                 removeOne(&queuedRequests,prop);
 174                 if (contains(activeRequests,prop))
 175                 {
 176                         return;
 177                 }
 178                 activeRequests.push_back(prop);
 179
 180                 QVariantMap reply;
 181
 182                 reply["type"] = "method";
 183                 reply["name"] = "subscribe";
 184                 reply["property"] = prop.c_str();
 185                 reply["transactionid"] = "d293f670-f0b3-11e1-aff1-0800200c9a66";
 186
 187                 lwsWriteVariant(clientsocket, reply);
 188         }
 189 }
 190 void WebSocketSource::setConfiguration(map<string, string> config)
 191 {
 192         //printf("WebSocketSource::setConfiguration has been called\n");
 193         std::string ip;
 194         int port;
 195         configuration = config;
 196
 197         if(config.find("binaryProtocol") != config.end())
 198         {
 199                 doBinary = config["binaryProtocol"] == "true";
 200         }
 201
 202         for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
 203         {
 204                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting for WebSocketSource:" << (*i).first << ":" << (*i).second << "\n";
 205                 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
 206                 if ((*i).first == "ip")
 207                 {
 208                         ip = (*i).second;
 209                 }
 210                 if ((*i).first == "port")
 211                 {
 212                         port = boost::lexical_cast<int>((*i).second);
 213                 }
 214                 if ((*i).first == "ssl")
 215                 {
 216                         if ((*i).second == "true")
 217                         {
 218                                 m_sslEnabled = true;
 219                         }
 220                         else
 221                         {
 222                                 m_sslEnabled = false;
 223                         }
 224                 }
 225         }
 226         //printf("Connecting to websocket server at %s port %i\n",ip.c_str(),port);
 227         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Connecting to websocket server at" << ip << ":" << port << "\n";
 228         int sslval = 0;
 229         if (m_sslEnabled)
 230         {
 231                 DebugOut(5) << "SSL ENABLED" << endl;
 232                 sslval = 2;
 233         }
 234
 235         clientsocket = libwebsocket_client_connect(context, ip.c_str(), port, sslval,"/", "localhost", "websocket", protocols[0].name, -1);
 236 }
 237
 238 PropertyInfo WebSocketSource::getPropertyInfo(const VehicleProperty::Property &property)
 239 {
 240         Zone::ZoneList zones;
 241         for(auto i : properties.properties())
 242         {
 243                 if(i->name == property)
 244                 {
 245                         zones.push_back(i->zone);
 246                 }
 247         }
 248
 249         return PropertyInfo(0, zones);
 250 }
 251
 252 bool gioPollingFunc(GIOChannel *source, GIOCondition condition, gpointer data)
 253 {
 254         //This is the polling function. If it return false, glib will stop polling this FD.
 255
 256         oldTimestamp = amb::currentTime();
 257
 258         struct pollfd pollstruct;
 259         int newfd = g_io_channel_unix_get_fd(source);
 260         pollstruct.fd = newfd;
 261         pollstruct.events = condition;
 262         pollstruct.revents = condition;
 263         libwebsocket_service_fd(context,&pollstruct);
 264         if (condition & G_IO_HUP)
 265         {
 266                 //Hang up. Returning false closes out the GIOChannel.
 267                 //printf("Callback on G_IO_HUP\n");
 268                 return false;
 269         }
 270         if (condition & G_IO_IN)
 271         {
 272
 273         }
 274         DebugOut() << "gioPollingFunc" << condition << endl;
 275
 276         return true;
 277 }
 278
 279 static int checkTimeouts(gpointer data)
 280 {
 281         WebSocketSource *src = (WebSocketSource*)data;
 282         for (auto i=src->uuidTimeoutMap.begin();i!= src->uuidTimeoutMap.end();i++)
 283         {
 284                 if (src->uuidRangedReplyMap.find((*i).first) != src->uuidRangedReplyMap.end())
 285                 {
 286                         //A source exists!
 287                         if (amb::currentTime() > (*i).second)
 288                         {
 289                                 //We've reached timeout
 290                                 DebugOut() << "Timeout reached for request ID:" << (*i).first << "\n";
 291                                 src->uuidRangedReplyMap[(*i).first]->success = false;
 292                                 src->uuidRangedReplyMap[(*i).first]->completed(src->uuidRangedReplyMap[(*i).first]);
 293                                 src->uuidRangedReplyMap.erase((*i).first);
 294                                 src->uuidTimeoutMap.erase((*i).first);
 295                                 i--;
 296
 297                                 if (src->uuidTimeoutMap.size() == 0)
 298                                 {
 299                                         return 0;
 300                                 }
 301
 302                         }
 303                         else
 304                         {
 305                                 //No timeout yet, keep waiting.
 306                         }
 307                 }
 308                 else
 309                 {
 310                         //Reply has already come back, ignore and erase from list.
 311                         src->uuidTimeoutMap.erase((*i).first);
 312                         i--;
 313
 314                         if (src->uuidTimeoutMap.size() == 0)
 315                         {
 316                                 return 0;
 317                         }
 318                 }
 319
 320         }
 321         return 0;
 322 }
 323
 324 static int callback_http_only(libwebsocket_context *context, struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user, void *in, size_t len)
 325 {
 326         unsigned char buf[LWS_SEND_BUFFER_PRE_PADDING + 4096 + LWS_SEND_BUFFER_POST_PADDING];
 327         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << reason << "callback_http_only" << endl;
 328         switch (reason)
 329         {
 330                 case LWS_CALLBACK_CLOSED:
 331                         //fprintf(stderr, "mirror: LWS_CALLBACK_CLOSED\n");
 332                         //wsi_mirror = NULL;
 333                         //printf("Connection closed!\n");
 334                         break;
 335
 336                 //case LWS_CALLBACK_PROTOCOL_INIT:
 337                 case LWS_CALLBACK_CLIENT_ESTABLISHED:
 338                 {
 339                         //This happens when a client initally connects. We need to request the support event types.
 340                         source->clientConnected = true;
 341                         source->checkSubscriptions();
 342                         //printf("Incoming connection!\n");
 343                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming connection" << endl;
 344
 345                         QVariantMap toSend;
 346                         toSend["type"] = "method";
 347                         toSend["name"] = "getSupported";
 348                         toSend["transactionid"] = amb::createUuid().c_str();
 349
 350                         lwsWriteVariant(wsi, toSend);
 351
 352                         break;
 353                 }
 354                 case LWS_CALLBACK_CLIENT_RECEIVE:
 355                 {
 356                         QByteArray d((char*)in, len);
 357
 358                         WebSocketSource * manager = source;
 359
 360                         if(manager->expectedMessageFrames && manager->partialMessageIndex < manager->expectedMessageFrames)
 361                         {
 362                                 manager->incompleteMessage += d;
 363                                 manager->partialMessageIndex++;
 364                                 break;
 365                         }
 366                         else if(manager->expectedMessageFrames && manager->partialMessageIndex == manager->expectedMessageFrames)
 367                         {
 368                                 d = manager->incompleteMessage + d;
 369                                 manager->expectedMessageFrames = 0;
 370                         }
 371
 372                         QJsonDocument doc;
 373
 374                         if(doBinary)
 375                                 doc = QJsonDocument::fromBinaryData(d);
 376                         else
 377                         {
 378                                 doc = QJsonDocument::fromJson(d);
 379                                 DebugOut(7)<<d.data()<<endl;
 380                         }
 381
 382                         if(doc.isNull())
 383                         {
 384                                 DebugOut(DebugOut::Warning)<<"Invalid message"<<endl;
 385                                 break;
 386                         }
 387
 388                         QVariantMap call = doc.toVariant().toMap();
 389
 390                         string type = call["type"].toString().toStdString();
 391                         string name = call["name"].toString().toStdString();
 392                         string id = call["transactionid"].toString().toStdString();
 393
 394                         if(type == "multiframe")
 395                         {
 396                                 manager->expectedMessageFrames = call["frames"].toInt();
 397                                 manager->partialMessageIndex = 1;
 398                                 manager->incompleteMessage = "";
 399
 400                         }
 401                         else if (type == "valuechanged")
 402                         {
 403                                 QVariantMap data = call["data"].toMap();
 404
 405                                 string value = data["value"].toString().toStdString();
 406                                 double timestamp = data["timestamp"].toDouble();
 407                                 int sequence = data["sequence"].toInt();
 408                                 Zone::Type zone = data["zone"].toInt();
 409                                 string type = data["type"].toString().toStdString();
 410
 411                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Value changed:" << name << value << endl;
 412
 413                                 try
 414                                 {
 415                                         auto property = properties.append(name, source->uuid(), zone, type);
 416
 417                                         if(!property)
 418                                         {
 419                                                 DebugOut(DebugOut::Warning) << "We either don't have this or don't support it ("
 420                                                                                                         << name << "," << zone << "," << type << ")" << endl;
 421                                         }
 422
 423                                         property->timestamp = timestamp;
 424                                         property->sequence = sequence;
 425                                         property->fromString(value);
 426
 427                                         m_re->updateProperty(property.get(), source->uuid());
 428
 429                                         double currenttime = amb::currentTime();
 430
 431                                         /** This is now the latency between when something is available to read on the socket, until
 432                                          *  a property is about to be updated in AMB.  This includes libwebsockets parsing and the
 433                                          *  JSON parsing in this section.
 434                                          */
 435
 436                                         DebugOut()<<"websocket network + parse latency: "<<(currenttime - property->timestamp)*1000<<"ms"<<endl;
 437                                         totalTime += (currenttime - oldTimestamp)*1000;
 438                                         numUpdates ++;
 439                                         averageLatency = totalTime / numUpdates;
 440
 441                                         DebugOut()<<"Average parse latency: "<<averageLatency<<endl;
 442                                 }
 443                                 catch (exception ex)
 444                                 {
 445                                         //printf("Exception %s\n",ex.what());
 446                                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Exception:" << ex.what() << "\n";
 447                                 }
 448                         }
 449                         else if (type == "methodReply")
 450                         {
 451                                 if (name == "getSupported" || name == "supportedChanged")
 452                                 {
 453
 454                                         QVariant data = call["data"];
 455
 456                                         QVariantList supported = data.toList();
 457
 458                                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Got getSupported request"<<endl;
 459
 460
 461                                         Q_FOREACH(QVariant p, supported)
 462                                         {
 463                                                 QVariantMap d = p.toMap();
 464                                                 Zone::Type zone = d["zone"].toInt();
 465                                                 std::string name = d["property"].toString().toStdString();
 466                                                 std::string proptype = d["type"].toString().toStdString();
 467                                                 std::string source = d["source"].toString().toStdString();
 468
 469                                                 properties.append(name, source, zone, proptype);
 470                                         }
 471
 472                                         source->updateSupported();
 473
 474                                 }
 475                                 else if (name == "getRanged")
 476                                 {
 477                                         QVariantList data = call["data"].toList();
 478
 479                                         std::list<AbstractPropertyType*> propertylist;
 480
 481                                         Q_FOREACH(QVariant d, data)
 482                                         {
 483                                                 QVariantMap obj = d.toMap();
 484
 485                                                 std::string name = obj["property"].toString().toStdString();
 486                                                 std::string value = obj["value"].toString().toStdString();
 487                                                 double timestamp = obj["timestamp"].toDouble();
 488                                                 int sequence = obj["sequence"].toInt();
 489
 490                                                 AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(name, value);
 491                                                 if(!type)
 492                                                 {
 493                                                         DebugOut() << "TODO: support custom types here: " << endl;
 494                                                         continue;
 495                                                 }
 496                                                 type->timestamp = timestamp;
 497                                                 type->sequence = sequence;
 498
 499                                                 propertylist.push_back(type);
 500                                         }
 501
 502                                         if (source->uuidRangedReplyMap.find(id) != source->uuidRangedReplyMap.end())
 503                                         {
 504                                                 source->uuidRangedReplyMap[id]->values = propertylist;
 505                                                 source->uuidRangedReplyMap[id]->success = true;
 506                                                 source->uuidRangedReplyMap[id]->completed(source->uuidRangedReplyMap[id]);
 507                                                 source->uuidRangedReplyMap.erase(id);
 508                                         }
 509                                         else
 510                                         {
 511                                                 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";
 512                                         }
 513                                 }
 514                                 else if (name == "get")
 515                                 {
 516
 517                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "Got \"GET\" reply" << endl;
 518                                         if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end())
 519                                         {
 520                                                 QVariantMap obj = call["data"].toMap();
 521
 522                                                 std::string property = obj["property"].toString().toStdString();
 523                                                 std::string value = obj["value"].toString().toStdString();
 524                                                 double timestamp = obj["timestamp"].toDouble();
 525                                                 int sequence = obj["sequence"].toInt();
 526                                                 Zone::Type zone = obj["zone"].toInt();
 527
 528                                                 auto v = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(property, value));
 529
 530                                                 v->timestamp = timestamp;
 531                                                 v->sequence = sequence;
 532                                                 v->zone = zone;
 533
 534                                                 if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end() && source->uuidReplyMap[id]->error != AsyncPropertyReply::Timeout)
 535                                                 {
 536                                                         source->uuidReplyMap[id]->value = v.get();
 537                                                         source->uuidReplyMap[id]->success = true;
 538                                                         source->uuidReplyMap[id]->completed(source->uuidReplyMap[id]);
 539                                                         source->uuidReplyMap.erase(id);
 540
 541                                                 }
 542                                                 else
 543                                                 {
 544                                                         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";
 545                                                 }
 546                                         }
 547                                         else
 548                                         {
 549                                                 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "GET Method Reply INVALID! Multiple properties detected, only single are supported!!!" << "\n";
 550                                         }
 551
 552
 553                                 }
 554                                 else if (name == "set")
 555                                 {
 556                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "Got \"SET\" event" << endl;
 557                                         std::string id = call["transactionid"].toString().toStdString();
 558
 559                                         if(source->setReplyMap.find(id) != source->setReplyMap.end() && source->setReplyMap[id]->error != AsyncPropertyReply::Timeout)
 560                                         {
 561                                                 AsyncPropertyReply* reply = source->setReplyMap[id];
 562
 563                                                 reply->success = call["success"].toBool();
 564                                                 reply->error = AsyncPropertyReply::strToError(call["error"].toString().toStdString());
 565
 566                                                 QVariantMap obj = call["data"].toMap();
 567
 568                                                 std::string property = obj["property"].toString().toStdString();
 569                                                 std::string value = obj["value"].toString().toStdString();
 570
 571                                                 double timestamp = obj["timestamp"].toDouble();
 572                                                 int sequence = obj["sequence"].toInt();
 573                                                 Zone::Type zone = obj["zone"].toInt();
 574
 575                                                 auto v = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(property, value));
 576
 577                                                 if(v)
 578                                                 {
 579                                                         v->timestamp = timestamp;
 580                                                         v->sequence = sequence;
 581                                                         v->zone = zone;
 582                                                 }
 583                                                 else
 584                                                 {
 585                                                         throw std::runtime_error("property may not be registered.");
 586                                                 }
 587
 588                                                 reply->value = v.get();
 589                                                 reply->completed(reply);
 590                                                 source->setReplyMap.erase(id);
 591                                         }
 592                                 }
 593                                 else
 594                                 {
 595                                         DebugOut(DebugOut::Warning) << "Unhandled methodReply: " << name << endl;
 596                                 }
 597
 598                         }
 599
 600                         break;
 601
 602                 }
 603                 case LWS_CALLBACK_CLIENT_CONFIRM_EXTENSION_SUPPORTED:
 604                 {
 605                         //printf("Requested extension: %s\n",(char*)in);
 606                         return 0;
 607                         break;
 608                 }
 609                 case LWS_CALLBACK_ADD_POLL_FD:
 610                 {
 611                         DebugOut(5) << __SMALLFILE__ << ":" << __LINE__ << "Adding poll for websocket IO channel" << endl;
 612                         //Add a FD to the poll list.
 613                         GIOChannel *chan = g_io_channel_unix_new(libwebsocket_get_socket_fd(wsi));
 614
 615                         /// TODO: I changed this to be more consistent with the websocket sink end. it may not be correct. TEST
 616
 617                         g_io_add_watch(chan,GIOCondition(G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP),(GIOFunc)gioPollingFunc,0);
 618                         g_io_channel_set_close_on_unref(chan,true);
 619                         g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
 620
 621                         break;
 622                 }
 623                 return 0;
 624         }
 625 }
 626 void WebSocketSource::updateSupported()
 627 {
 628         PropertyList list;
 629         for(auto i : properties.properties())
 630         {
 631                 if(!contains(list, i->name))
 632                         list.push_back(i->name);
 633         }
 634
 635         m_re->updateSupported(list, PropertyList(), this);
 636 }
 637
 638 WebSocketSource::WebSocketSource(AbstractRoutingEngine *re, map<string, string> config) : AbstractSource(re, config), partialMessageIndex(0),expectedMessageFrames(0)
 639 {
 640         m_sslEnabled = false;
 641         clientConnected = false;
 642         source = this;
 643         m_re = re;
 644         struct lws_context_creation_info info;
 645         memset(&info, 0, sizeof info);
 646         info.protocols = protocols;
 647         info.extensions = nullptr;
 648
 649         if(config.find("useExtensions") != config.end() && config["useExtensions"] == "true")
 650         {
 651                 info.extensions = libwebsocket_get_internal_extensions();
 652         }
 653
 654         info.gid = -1;
 655         info.uid = -1;
 656         info.port = CONTEXT_PORT_NO_LISTEN;
 657         info.user = this;
 658
 659         context = libwebsocket_create_context(&info);
 660
 661         setConfiguration(config);
 662
 663         //printf("websocketsource loaded!!!\n");
 664         g_timeout_add(1000,checkTimeouts,this); //Do this once per second, check for functions that have timed out and reply with success = false;
 665
 666 }
 667 PropertyList WebSocketSource::supported()
 668 {
 669         PropertyList list;
 670         for(auto i : properties.properties())
 671         {
 672                 list.push_back(i->name);
 673         }
 674         return list;
 675 }
 676
 677 int WebSocketSource::supportedOperations()
 678 {
 679         /// TODO: need to do this correctly based on what the host supports.
 680         return Get | Set | GetRanged;
 681 }
 682
 683 const string WebSocketSource::uuid()
 684 {
 685         return "d293f670-f0b3-11e1-aff1-0800200c9a66";
 686 }
 687
 688 void WebSocketSource::subscribeToPropertyChanges(VehicleProperty::Property property)
 689 {
 690         //printf("Subscribed to property: %s\n",property.c_str());
 691         queuedRequests.push_back(property);
 692         if (clientConnected)
 693         {
 694                 checkSubscriptions();
 695         }
 696 }
 697
 698
 699 void WebSocketSource::unsubscribeToPropertyChanges(VehicleProperty::Property property)
 700 {
 701         removeRequests.push_back(property);
 702         if (clientConnected)
 703         {
 704                 checkSubscriptions();
 705         }
 706 }
 707
 708
 709 void WebSocketSource::getPropertyAsync(AsyncPropertyReply *reply)
 710 {
 711         std::string uuid = amb::createUuid();
 712         uuidReplyMap[uuid] = reply;
 713         uuidTimeoutMap[uuid] = amb::currentTime() + 10.0; ///TODO: 10 second timeout, make this configurable?
 714
 715         QVariantMap data;
 716         data["property"] = reply->property.c_str();
 717         data["zone"] = reply->zoneFilter;
 718
 719         QVariantMap replyvar;
 720         replyvar["type"] = "method";
 721         replyvar["name"] = "get";
 722         replyvar["data"] = data;
 723         replyvar["transactionid"] = uuid.c_str();
 724
 725         lwsWriteVariant(clientsocket, replyvar);
 726 }
 727
 728 void WebSocketSource::getRangePropertyAsync(AsyncRangePropertyReply *reply)
 729 {
 730         std::string uuid = amb::createUuid();
 731         uuidRangedReplyMap[uuid] = reply;
 732         uuidTimeoutMap[uuid] = amb::currentTime() + 60; ///TODO: 60 second timeout, make this configurable?
 733         stringstream s;
 734         s.precision(15);
 735         s << "{\"type\":\"method\",\"name\":\"getRanged\",\"data\": {";
 736
 737         QVariantMap replyvar;
 738         replyvar["type"] = "method";
 739         replyvar["name"] = "getRanged";
 740         replyvar["transactionid"] = uuid.c_str();
 741         replyvar["timeBegin"] = reply->timeBegin;
 742         replyvar["timeEnd"] = reply->timeEnd;
 743         replyvar["sequenceBegin"] = reply->sequenceBegin;
 744         replyvar["sequenceEnd"] = reply->sequenceEnd;
 745
 746         QStringList properties;
 747
 748         for (auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
 749         {
 750                 VehicleProperty::Property p = *itr;
 751                 properties.append(p.c_str());
 752         }
 753
 754         replyvar["data"] = properties;
 755
 756         lwsWriteVariant(clientsocket, replyvar);
 757 }
 758
 759 AsyncPropertyReply * WebSocketSource::setProperty( AsyncSetPropertyRequest request )
 760 {
 761         AsyncPropertyReply* reply = new AsyncPropertyReply(request);
 762
 763         std::string uuid = amb::createUuid();
 764
 765         QVariantMap data;
 766         data["property"] = request.property.c_str();
 767         data["value"] = request.value->toString().c_str();
 768         data["zone"] = request.zoneFilter;
 769
 770         QVariantMap replyvar;
 771         replyvar["type"] = "method";
 772         replyvar["name"] = "set";
 773         replyvar["data"] = data;
 774         replyvar["transactionid"] = uuid.c_str();
 775
 776         lwsWriteVariant(clientsocket, replyvar);
 777
 778         setReplyMap[uuid] = reply;
 779
 780         return reply;
 781 }
 782
 783 extern "C" AbstractSource * create(AbstractRoutingEngine* routingengine, map<string, string> config)
 784 {
 785         return new WebSocketSource(routingengine, config);
 786
 787 }