plugins/websocketsink/websocketsinkmanager.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 "websocketsinkmanager.h"
  21 #include "websocketsink.h"
  22 #include <sstream>
  23 #include <json-glib/json-glib.h>
  24 #include <listplusplus.h>
  25 #define __SMALLFILE__ std::string(__FILE__).substr(std::string(__FILE__).rfind("/")+1)
  26
  27 //Global variables, these will be moved into the class
  28 struct pollfd pollfds[100];
  29 int count_pollfds = 0;
  30 libwebsocket_context *context;
  31 WebSocketSinkManager *sinkManager;
  32 static int websocket_callback(struct libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user,void *in, size_t len);
  33 bool gioPollingFunc(GIOChannel *source,GIOCondition condition,gpointer data);
  34
  35
  36
  37 WebSocketSinkManager::WebSocketSinkManager(AbstractRoutingEngine* engine, map<string, string> config):AbstractSinkManager(engine, config)
  38 {
  39         m_engine = engine;
  40
  41
  42         //Create a listening socket on port 23000 on localhost.
  43
  44
  45 }
  46 void WebSocketSinkManager::init()
  47 {
  48         //Protocol list for libwebsockets.
  49         protocollist[0] = { "http-only", websocket_callback, 0 };
  50         protocollist[1] = { NULL, NULL, 0 };
  51
  52
  53         setConfiguration(configuration);
  54 }
  55 void WebSocketSinkManager::setConfiguration(map<string, string> config)
  56 {
  57 //      //Config has been passed, let's start stuff up.
  58         configuration = config;
  59
  60         //Default values
  61         int port = 23000;
  62         std::string interface = "lo";
  63         const char *ssl_cert_path = NULL;
  64         const char *ssl_key_path = NULL;
  65         int options = 0;
  66
  67         //Try to load config
  68         for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
  69         {
  70                 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
  71                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting:" << (*i).first << ":" << (*i).second << "\n";
  72                 if ((*i).first == "interface")
  73                 {
  74                         interface = (*i).second;
  75                 }
  76                 if ((*i).first == "port")
  77                 {
  78                         port = boost::lexical_cast<int>((*i).second);
  79                 }
  80         }
  81         context = libwebsocket_create_context(port, interface.c_str(), protocollist,libwebsocket_internal_extensions,ssl_cert_path, ssl_key_path, -1, -1, options);
  82 }
  83 void WebSocketSinkManager::addSingleShotSink(libwebsocket* socket, VehicleProperty::Property property,string id)
  84 {
  85         AsyncPropertyRequest velocityRequest;
  86         if (property == "running_status_speedometer")
  87         {
  88                 velocityRequest.property = VehicleProperty::VehicleSpeed;
  89         }
  90         else if (property == "running_status_engine_speed")
  91         {
  92                 velocityRequest.property = VehicleProperty::EngineSpeed;
  93         }
  94         else if (property == "running_status_steering_wheel_angle")
  95         {
  96                 velocityRequest.property = VehicleProperty::SteeringWheelAngle;
  97         }
  98         else if (property == "running_status_transmission_gear_status")
  99         {
 100                 velocityRequest.property = VehicleProperty::TransmissionShiftPosition;
 101         }
 102         else
 103         {
 104                 PropertyList foo = VehicleProperty::capabilities();
 105                 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(property))
 106                 {
 107                         velocityRequest.property = property;
 108                 }
 109                 else
 110                 {
 111                         DebugOut(0)<<"websocketsink: Invalid property requested: "<<property;
 112                         return;
 113                 }
 114
 115         }
 116         velocityRequest.completed = [socket,id,property](AsyncPropertyReply* reply)
 117         {
 118                 printf("Got property:%s\n",reply->value->toString().c_str());
 119                 //uint16_t velocity = boost::any_cast<uint16_t>(reply->value);
 120                 stringstream s;
 121
 122                 //TODO: Dirty hack hardcoded stuff, jsut to make it work.
 123                 string tmpstr = "";
 124                 tmpstr = property;
 125                 s << "{\"type\":\"methodReply\",\"name\":\"get\",\"data\":[{\"name\":\"" << tmpstr << "\",\"value\":\"" << reply->value->toString() << "\"}],\"transactionid\":\"" << id << "\"}";
 126
 127                 string replystr = s.str();
 128                 //printf("Reply: %s\n",replystr.c_str());
 129                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
 130
 131                 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
 132                 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
 133                 strcpy(new_response,replystr.c_str());
 134                 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
 135
 136                 //TODO: run valgrind on this. libwebsocket's documentation says NOTHING about this, yet malloc insists it's true.
 137                 //delete new_response; <- Unneeded. Apparently libwebsocket free's it.
 138                 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING); //Needs to subtract pre-padding, to get back to the start of the pointer.
 139
 140         };
 141
 142         AsyncPropertyReply* reply = routingEngine->getPropertyAsync(velocityRequest);
 143 }
 144 void WebSocketSinkManager::removeSink(libwebsocket* socket,VehicleProperty::Property property, string uuid)
 145 {
 146         if (m_sinkMap.find(property) != m_sinkMap.end())
 147         {
 148                 list<WebSocketSink*> sinks = m_sinkMap[property];
 149
 150                 for(auto i = sinks.begin(); i != sinks.end(); i++)
 151                 {
 152                         delete *i;
 153                 }
 154
 155                 m_sinkMap.erase(property);
 156
 157                 stringstream s;
 158                 s << "{\"type\":\"methodReply\",\"name\":\"unsubscribe\",\"data\":[\"" << property << "\"],\"transactionid\":\"" << uuid << "\"}";
 159
 160                 string replystr = s.str();
 161                 //printf("Reply: %s\n",replystr.c_str());
 162                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
 163
 164                 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
 165                 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
 166                 strcpy(new_response,replystr.c_str());
 167                 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
 168                 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
 169         }
 170 }
 171 void WebSocketSinkManager::setValue(string property,string value)
 172 {
 173         AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(property,value);
 174
 175         AsyncSetPropertyRequest request;
 176         request.property = property;
 177         request.value = type;
 178         request.completed = [](AsyncPropertyReply* reply)
 179         {
 180                 ///TODO: do something here on !reply->success
 181                 delete reply;
 182         };
 183
 184         m_engine->setProperty(request);
 185         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "AbstractRoutingEngine::setProperty called with arguments:" << property << value << "\n";
 186         delete type;
 187
 188 }
 189 void WebSocketSinkManager::addSink(libwebsocket* socket, VehicleProperty::Property property,string uuid)
 190 {
 191         stringstream s;
 192
 193         //TODO: Dirty hack hardcoded stuff, jsut to make it work.
 194         string tmpstr = "";
 195         if (property == "running_status_speedometer")
 196         {
 197                 tmpstr = VehicleProperty::VehicleSpeed;
 198         }
 199         else if (property == "running_status_engine_speed")
 200         {
 201                 tmpstr = VehicleProperty::EngineSpeed;
 202         }
 203         else if (property == "running_status_steering_wheel_angle")
 204         {
 205                 tmpstr = VehicleProperty::SteeringWheelAngle;
 206         }
 207         else if (property == "running_status_transmission_gear_status")
 208         {
 209                 tmpstr = VehicleProperty::TransmissionShiftPosition;
 210         }
 211         else
 212         {
 213                 PropertyList foo = VehicleProperty::capabilities();
 214                 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(property))
 215                 {
 216                         tmpstr = property;
 217                 }
 218                 else
 219                 {
 220                         //Invalid property requested.
 221                         return;
 222                 }
 223
 224         }
 225         s << "{\"type\":\"methodReply\",\"name\":\"subscribe\",\"data\":[\"" << property << "\"],\"transactionid\":\"" << uuid << "\"}";
 226
 227         string replystr = s.str();
 228         //printf("Reply: %s\n",replystr.c_str());
 229         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
 230
 231         char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
 232         new_response+=LWS_SEND_BUFFER_PRE_PADDING;
 233         strcpy(new_response,replystr.c_str());
 234         libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
 235         delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
 236         WebSocketSink *sink = new WebSocketSink(m_engine,socket,uuid,property,tmpstr);
 237         m_sinkMap[property].push_back(sink);
 238 }
 239 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
 240 {
 241         sinkManager = new WebSocketSinkManager(routingengine, config);
 242         sinkManager->init();
 243         return sinkManager;
 244 }
 245 void WebSocketSinkManager::disconnectAll(libwebsocket* socket)
 246 {
 247         std::list<WebSocketSink*> toDeleteList;
 248
 249         for (auto i=m_sinkMap.begin(); i != m_sinkMap.end();i++)
 250         {
 251                 std::list<WebSocketSink*> *sinks = & (*i).second;
 252                 for (auto sinkItr =  sinks->begin(); sinkItr != sinks->end(); sinkItr++)
 253                 {
 254                         if ((*sinkItr)->socket() == socket)
 255                         {
 256                                 //This is the sink in question.
 257                                 WebSocketSink* sink = (*sinkItr);
 258                                 if(!ListPlusPlus<WebSocketSink*>(&toDeleteList).contains(sink))
 259                                 {
 260                                         toDeleteList.push_back(sink);
 261                                 }
 262
 263                                 sinks->erase(sinkItr);
 264                                 sinkItr = sinks->begin();
 265                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Sink removed"<<endl;
 266                         }
 267                 }
 268         }
 269
 270         for(auto i=toDeleteList.begin();i!=toDeleteList.end();i++)
 271         {
 272                 delete *i;
 273         }
 274 }
 275 void WebSocketSinkManager::addPoll(int fd)
 276 {
 277         GIOChannel *chan = g_io_channel_unix_new(fd);
 278         guint sourceid = g_io_add_watch(chan,G_IO_IN,(GIOFunc)gioPollingFunc,chan);
 279         g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
 280         m_ioChannelMap[fd] = chan;
 281         m_ioSourceMap[fd] = sourceid;
 282 }
 283 void WebSocketSinkManager::removePoll(int fd)
 284 {
 285         g_io_channel_shutdown(m_ioChannelMap[fd],false,0);
 286         //printf("Shutting down IO Channel\n");
 287         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Shutting down IO Channel\n";
 288         g_source_remove(m_ioSourceMap[fd]); //Since the watch owns the GIOChannel, this should unref it enough to dissapear.
 289         //for (map<int,guint>::const_iterator i=m_ioSourceMap.cbegin();i!=m_ioSourceMap.cend();i++)
 290         for (map<int,guint>::iterator i=m_ioSourceMap.begin();i!=m_ioSourceMap.end();i++)
 291         {
 292                 if((*i).first == fd)
 293                 {
 294                         //printf("Erasing source\n");
 295                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing source\n";
 296                         m_ioSourceMap.erase(i);
 297                         i--;
 298                 }
 299         }
 300         //for (map<int,GIOChannel*>::const_iterator i=m_ioChannelMap.cbegin();i!=m_ioChannelMap.cend();i++)
 301         for (map<int,GIOChannel*>::iterator i=m_ioChannelMap.begin();i!=m_ioChannelMap.end();i++)
 302         {
 303                 if((*i).first == fd)
 304                 {
 305                         //printf("Erasing channel\n");
 306                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing channel\n";
 307                         m_ioChannelMap.erase(i);
 308                         i--;
 309                 }
 310         }
 311 }
 312
 313 static int websocket_callback(struct libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user,void *in, size_t len)
 314 {
 315         //printf("Switch: %i\n",reason);
 316
 317
 318         switch (reason)
 319         {
 320                 case LWS_CALLBACK_CLIENT_WRITEABLE:
 321                 {
 322                         //Connection has been established.
 323                         //printf("Connection established\n");
 324                         break;
 325                 }
 326                 case LWS_CALLBACK_CLOSED:
 327                 {
 328                         //Connection is closed, we need to remove all related sinks
 329                         sinkManager->disconnectAll(wsi);
 330                         /*g_io_
 331                         GIOChannel *chan = g_io_channel_unix_new((int)(long)user);
 332                         g_io_add_watch(chan,G_IO_IN,(GIOFunc)gioPollingFunc,0);
 333                         g_io_add_watch(chan,G_IO_PRI,(GIOFunc)gioPollingFunc,0);
 334                         pollfds[count_pollfds].fd = (int)(long)user;
 335                         pollfds[count_pollfds].events = (int)len;
 336 //                      pollfds[count_pollfds++].revents = 0;*/
 337                         break;
 338                 }
 339                 case LWS_CALLBACK_CLIENT_RECEIVE:
 340                 {
 341                         //printf("Client writable\n");
 342                         break;
 343                 }
 344                 case LWS_CALLBACK_SERVER_WRITEABLE:
 345                 {
 346                         //printf("Server writable\n");
 347                         break;
 348                 }
 349
 350                 case LWS_CALLBACK_RECEIVE:
 351                 {
 352                         //printf("Data Received: %s\n",(char*)in);
 353                         //The lack of a break; here is intentional.
 354                 }
 355                 case LWS_CALLBACK_HTTP:
 356                 {
 357                         //TODO: Verify that ALL requests get sent via LWS_CALLBACK_HTTP, so we can use that instead of LWS_CALLBACK_RECIEVE
 358                         //TODO: Do we want exceptions, or just to return an invalid json reply? Probably an invalid json reply.
 359                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " Requested: " << (char*)in << "\n";
 360                         GError* error = nullptr;
 361
 362
 363                         JsonParser* parser = json_parser_new();
 364                         if (!json_parser_load_from_data(parser,(char*)in,len,&error))
 365                         {
 366                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Error loading JSON\n";
 367                                 return 0;
 368                         }
 369
 370                         JsonNode* node = json_parser_get_root(parser);
 371                         if(node == nullptr)
 372                         {
 373                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Error getting root node of json\n";
 374                                 //throw std::runtime_error("Unable to get JSON root object");
 375                                 return 0;
 376                         }
 377
 378                         JsonReader* reader = json_reader_new(node);
 379                         if(reader == nullptr)
 380                         {
 381                                 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "json_reader is null!\n";
 382                                 //throw std::runtime_error("Unable to create JSON reader");
 383                                 return 0;
 384                         }
 385
 386
 387
 388
 389
 390                         string type;
 391                         json_reader_read_member(reader,"type");
 392                         type = json_reader_get_string_value(reader);
 393                         json_reader_end_member(reader);
 394
 395                         string  name;
 396                         json_reader_read_member(reader,"name");
 397                         name = json_reader_get_string_value(reader);
 398                         json_reader_end_member(reader);
 399
 400                         list<string> data;
 401                         list<string> key;
 402                         list<string> value;
 403                         json_reader_read_member(reader,"data");
 404                         if (json_reader_is_array(reader))
 405                         {
 406                                 for(int i=0; i < json_reader_count_elements(reader); i++)
 407                                 {
 408                                         json_reader_read_element(reader,i);
 409                                         if (json_reader_is_value(reader))
 410                                         {
 411                                                 //Raw string value
 412                                                 string path = json_reader_get_string_value(reader);
 413                                                 data.push_back(path);
 414
 415                                         }
 416                                         else
 417                                         {
 418                                                 //Not a raw string value, then it's "property/value" kvp, for "set" requests
 419                                                 json_reader_read_member(reader,"property");
 420                                                 string keystr = json_reader_get_string_value(reader);
 421                                                 key.push_back(keystr);
 422                                                 json_reader_end_member(reader);
 423                                                 json_reader_read_member(reader,"value");
 424                                                 string valuestr = json_reader_get_string_value(reader);
 425                                                 value.push_back(valuestr);
 426                                                 json_reader_end_member(reader);
 427                                         }
 428                                         json_reader_end_element(reader);
 429                                 }
 430                         }
 431                         else
 432                         {
 433                                 string path = json_reader_get_string_value(reader);
 434                                 if (path != "")
 435                                 {
 436                                         data.push_back(path);
 437                                 }
 438                         }
 439                         json_reader_end_member(reader);
 440
 441                         string id;
 442                         json_reader_read_member(reader,"transactionid");
 443                         if (strcmp("gchararray",g_type_name(json_node_get_value_type(json_reader_get_value(reader)))) == 0)
 444                         {
 445                                 //Type is a string
 446                                 id = json_reader_get_string_value(reader);
 447                         }
 448                         else
 449                         {
 450                                 //Type is an integer
 451                                 stringstream strstr;
 452                                 strstr << json_reader_get_int_value(reader);
 453                                 id = strstr.str();
 454                         }
 455                         json_reader_end_member(reader);
 456
 457                         ///TODO: this will probably explode:
 458                         //mlc: I agree with Kevron here, it does explode.
 459                         //if(error) g_error_free(error);
 460
 461                         g_object_unref(reader);
 462                         g_object_unref(parser);
 463
 464
 465                         if (type == "method")
 466                         {
 467                                 if (name == "get")
 468                                 {
 469                                         if (data.size() > 0)
 470                                         {
 471                                                 //GetProperty is going to be a singleshot sink.
 472                                                 //string arg = arguments.front();
 473                                                 sinkManager->addSingleShotSink(wsi,data.front(),id);
 474                                                 /*if (data.front()== "running_status_speedometer")
 475                                                 {
 476                                                         sinkManager->addSingleShotSink(wsi,VehicleProperty::VehicleSpeed,id);
 477                                                 }
 478                                                 else if (data.front() == "running_status_engine_speed")
 479                                                 {
 480                                                         sinkManager->addSingleShotSink(wsi,VehicleProperty::EngineSpeed,id);
 481                                                 }
 482                                                 else if (data.front() == "running_status_steering_wheel_angle")
 483                                                 {
 484                                                         sinkManager->addSingleShotSink(wsi,VehicleProperty::SteeringWheelAngle,id);
 485                                                 }
 486                                                 else if (data.front() == "running_status_transmission_gear_status")
 487                                                 {
 488                                                         sinkManager->addSingleShotSink(wsi,VehicleProperty::TransmissionShiftPosition,id);
 489                                                 }
 490                                                 else
 491                                                 {
 492                                                   PropertyList foo = VehicleProperty::capabilities();
 493                                                   if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(data.front()))
 494                                                   {
 495                                                     sinkManager->addSingleShotSink(wsi,data.front(),id);
 496                                                   }
 497                                                 }*/
 498                                         }
 499                                         else
 500                                         {
 501                                                 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " \"get\" method called with no data! Transaction ID:" << id << "\n";
 502                                         }
 503                                 }
 504                                 else if (name == "set")
 505                                 {
 506                                         if (data.size() > 0)
 507                                         {
 508                                           //Should not happen
 509                                         }
 510                                         else if (value.size() > 0)
 511                                         {
 512                                                 if (key.size() != value.size())
 513                                                 {
 514                                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "\"set\" method called with an invalid key value pair count\n";
 515                                                 }
 516                                                 else
 517                                                 {
 518                                                         list<string>::iterator d = value.begin();
 519                                                         for (list<string>::iterator i=key.begin();i!=key.end();i++)
 520                                                         {
 521                                                                 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "websocketsinkmanager setting" << (*i) << "to" << (*d) << "\n";
 522                                                                 //(*i);
 523                                                                 sinkManager->setValue((*i),(*d));
 524                                                                 //(*d);
 525                                                                 d++;
 526                                                         }
 527
 528                                                 }
 529                                         }
 530                                 }
 531                                 else if (name == "subscribe")
 532                                 {
 533                                         //Websocket wants to subscribe to an event, data.front();
 534                                         for (list<string>::iterator i=data.begin();i!=data.end();i++)
 535                                         {
 536                                                 sinkManager->addSink(wsi,(*i),id);
 537                                         }
 538                                 }
 539                                 else if (name == "unsubscribe")
 540                                 {
 541                                         //Websocket wants to unsubscribe to an event, data.front();
 542                                         for (list<string>::iterator i=data.begin();i!=data.end();i++)
 543                                         {
 544                                                 sinkManager->removeSink(wsi,(*i),id);
 545                                         }
 546                                 }
 547                                 else if (name == "getSupportedEventTypes")
 548                                 {
 549                                         //If data.front() dosen't contain a property name, return a list of properties supported.
 550                                         //if it does, then return the event types that particular property supports.
 551                                         string typessupported = "";
 552                                         if (data.size() == 0)
 553                                         {
 554                                                 //Send what properties we support
 555                                                 typessupported = "\"running_status_speedometer\",\"running_status_engine_speed\",\"running_status_steering_wheel_angle\",\"running_status_transmission_gear_status\"";
 556                                                 PropertyList foo = VehicleProperty::capabilities();
 557                                                 PropertyList::const_iterator i=foo.cbegin();
 558                                                 while (i != foo.cend())
 559                                                 {
 560                                                         typessupported.append(",\"").append((*i)).append("\"");
 561                                                         i++;
 562                                                 }
 563                                         }
 564                                         else
 565                                         {
 566                                                 //Send what events a particular property supports
 567                                                 if (data.front()== "running_status_speedometer")
 568                                                 {
 569                                                         typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
 570                                                 }
 571                                                 else if (data.front()== "running_status_engine_speed")
 572                                                 {
 573                                                         typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
 574                                                 }
 575                                                 else if (data.front() == "running_status_steering_wheel_angle")
 576                                                 {
 577                                                         typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
 578                                                 }
 579                                                 else if (data.front() == "running_status_transmission_gear_status")
 580                                                 {
 581                                                         typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
 582                                                 }
 583                                                 else
 584                                                 {
 585                                                         PropertyList foo = VehicleProperty::capabilities();
 586                                                         if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(data.front()))
 587                                                         {
 588                                                                 //sinkManager->addSingleShotSink(wsi,data.front(),id);
 589                                                                 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
 590                                                         }
 591                                                 }
 592                                         }
 593                                         stringstream s;
 594                                         string s2;
 595                                         s << "{\"type\":\"methodReply\",\"name\":\"getSupportedEventTypes\",\"data\":[" << typessupported << "],\"transactionid\":\"" << id << "\"}";
 596                                         string replystr = s.str();
 597                                         DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " JSON Reply: " << replystr << "\n";
 598                                         //printf("Reply: %s\n",replystr.c_str());
 599                                         char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
 600                                         new_response+=LWS_SEND_BUFFER_PRE_PADDING;
 601                                         strcpy(new_response,replystr.c_str());
 602                                         libwebsocket_write(wsi, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
 603                                         delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
 604                                 }
 605                         }
 606                         break;
 607                 }
 608                 case LWS_CALLBACK_ADD_POLL_FD:
 609                 {
 610                         //printf("Adding poll %i\n",sinkManager);
 611                         //DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Adding poll" << (int)sinkManager << "\n";
 612                         if (sinkManager != 0)
 613                         {
 614                                 sinkManager->addPoll((int)(long)user);
 615                         }
 616                         break;
 617                 }
 618                 case LWS_CALLBACK_DEL_POLL_FD:
 619                 {
 620                         sinkManager->removePoll((int)(long)user);
 621                         break;
 622                 }
 623                 case LWS_CALLBACK_SET_MODE_POLL_FD:
 624                 {
 625                         //Set the poll mode
 626                         break;
 627                 }
 628                 case LWS_CALLBACK_CLEAR_MODE_POLL_FD:
 629                 {
 630                         //Don't handle this yet.
 631                         break;
 632                 }
 633                 default:
 634                 {
 635                         //printf("Unhandled callback: %i\n",reason);
 636                         DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Unhandled callback:" << reason << "\n";
 637                         break;
 638                 }
 639         }
 640         return 0;
 641 }
 642
 643 bool gioPollingFunc(GIOChannel *source,GIOCondition condition,gpointer data)
 644 {
 645         if (condition != G_IO_IN)
 646         {
 647                 //Don't need to do anything
 648                 if (condition == G_IO_HUP)
 649                 {
 650                         //Hang up. Returning false closes out the GIOChannel.
 651                         //printf("Callback on G_IO_HUP\n");
 652                         return false;
 653                 }
 654                 return true;
 655         }
 656         //This is the polling function. If it return false, glib will stop polling this FD.
 657         //printf("Polling...%i\n",condition);
 658         lws_tokens token;
 659         struct pollfd pollstruct;
 660         int newfd = g_io_channel_unix_get_fd(source);
 661         pollstruct.fd = newfd;
 662         pollstruct.events = condition;
 663         pollstruct.revents = condition;
 664         libwebsocket_service_fd(context,&pollstruct);
 665         return true;
 666 }
 667