2 Copyright (C) 2012 Intel Corporation
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.
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.
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
20 #include "websocketsinkmanager.h"
21 #include "websocketsink.h"
23 #include <json-glib/json-glib.h>
24 #include <listplusplus.h>
25 #define __SMALLFILE__ std::string(__FILE__).substr(std::string(__FILE__).rfind("/")+1)
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);
37 WebSocketSinkManager::WebSocketSinkManager(AbstractRoutingEngine* engine, map<string, string> config):AbstractSinkManager(engine, config)
42 //Create a listening socket on port 23000 on localhost.
46 void WebSocketSinkManager::init()
48 //Protocol list for libwebsockets.
49 protocollist[0] = { "http-only", websocket_callback, 0 };
50 protocollist[1] = { NULL, NULL, 0 };
53 setConfiguration(configuration);
55 list< VehicleProperty::Property > WebSocketSinkManager::getSupportedProperties()
57 return m_engine->supported();
59 void WebSocketSinkManager::setConfiguration(map<string, string> config)
61 // //Config has been passed, let's start stuff up.
62 configuration = config;
66 std::string interface = "lo";
67 const char *ssl_cert_path = NULL;
68 const char *ssl_key_path = NULL;
72 for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
74 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
75 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting:" << (*i).first << ":" << (*i).second << "\n";
76 if ((*i).first == "interface")
78 interface = (*i).second;
80 if ((*i).first == "port")
82 port = boost::lexical_cast<int>((*i).second);
85 context = libwebsocket_create_context(port, interface.c_str(), protocollist,libwebsocket_internal_extensions,ssl_cert_path, ssl_key_path, -1, -1, options);
88 void WebSocketSinkManager::addSingleShotSink(libwebsocket* socket, VehicleProperty::Property property,string id)
90 AsyncPropertyRequest velocityRequest;
91 if (property == "running_status_speedometer")
93 velocityRequest.property = VehicleProperty::VehicleSpeed;
95 else if (property == "running_status_engine_speed")
97 velocityRequest.property = VehicleProperty::EngineSpeed;
99 else if (property == "running_status_steering_wheel_angle")
101 velocityRequest.property = VehicleProperty::SteeringWheelAngle;
103 else if (property == "running_status_transmission_gear_status")
105 velocityRequest.property = VehicleProperty::TransmissionShiftPosition;
109 PropertyList foo = VehicleProperty::capabilities();
110 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(property))
112 velocityRequest.property = property;
116 DebugOut(0)<<"websocketsink: Invalid property requested: "<<property;
121 velocityRequest.completed = [socket,id,property](AsyncPropertyReply* reply)
123 printf("Got property:%s\n",reply->value->toString().c_str());
124 //uint16_t velocity = boost::any_cast<uint16_t>(reply->value);
127 //TODO: Dirty hack hardcoded stuff, jsut to make it work.
130 s << "{\"type\":\"methodReply\",\"name\":\"get\",\"data\":[{\"name\":\"" << tmpstr << "\",\"value\":\"" << reply->value->toString() << "\"}],\"transactionid\":\"" << id << "\"}";
132 string replystr = s.str();
133 //printf("Reply: %s\n",replystr.c_str());
134 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
136 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
137 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
138 strcpy(new_response,replystr.c_str());
139 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
141 //TODO: run valgrind on this. libwebsocket's documentation says NOTHING about this, yet malloc insists it's true.
142 //delete new_response; <- Unneeded. Apparently libwebsocket free's it.
143 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING); //Needs to subtract pre-padding, to get back to the start of the pointer.
147 AsyncPropertyReply* reply = routingEngine->getPropertyAsync(velocityRequest);
150 void WebSocketSinkManager::addSingleShotRangedSink(libwebsocket* socket, VehicleProperty::Property property, double start, double end, string id)
152 AsyncRangePropertyRequest rangedRequest;
154 rangedRequest.begin = start;
155 rangedRequest.end = end;
157 if (property == "running_status_speedometer")
159 rangedRequest.property = VehicleProperty::VehicleSpeed;
161 else if (property == "running_status_engine_speed")
163 rangedRequest.property = VehicleProperty::EngineSpeed;
165 else if (property == "running_status_steering_wheel_angle")
167 rangedRequest.property = VehicleProperty::SteeringWheelAngle;
169 else if (property == "running_status_transmission_gear_status")
171 rangedRequest.property = VehicleProperty::TransmissionShiftPosition;
175 PropertyList foo = VehicleProperty::capabilities();
176 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(property))
178 rangedRequest.property = property;
182 DebugOut(0)<<"websocketsink: Invalid property requested: "<<property;
187 rangedRequest.completed = [socket,id](AsyncRangePropertyReply* reply)
191 //TODO: Dirty hack hardcoded stuff, jsut to make it work.
192 stringstream data ("[");
193 std::list<AbstractPropertyType*> values = reply->values;
194 for(auto itr = values.begin(); itr != values.end(); itr++)
196 if(itr != values.begin())
201 data << "{ \"value\" : " << "\"" << (*itr)->toString() << "\", \"time\" : \"" << (*itr)->timestamp << "\" }";
206 s << "{\"type\":\"methodReply\",\"name\":\"getHistory\",\"data\":"<<data<<",\"transactionid\":\"" << id << "\"}";
208 string replystr = s.str();
209 //printf("Reply: %s\n",replystr.c_str());
210 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
212 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
213 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
214 strcpy(new_response,replystr.c_str());
215 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
217 //TODO: run valgrind on this. libwebsocket's documentation says NOTHING about this, yet malloc insists it's true.
218 //delete new_response; <- Unneeded. Apparently libwebsocket free's it.
219 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING); //Needs to subtract pre-padding, to get back to the start of the pointer.
223 AsyncRangePropertyReply* reply = routingEngine->getRangePropertyAsync(rangedRequest);
226 void WebSocketSinkManager::removeSink(libwebsocket* socket,VehicleProperty::Property property, string uuid)
228 if (m_sinkMap.find(property) != m_sinkMap.end())
230 list<WebSocketSink*> sinks = m_sinkMap[property];
232 for(auto i = sinks.begin(); i != sinks.end(); i++)
237 m_sinkMap.erase(property);
240 s << "{\"type\":\"methodReply\",\"name\":\"unsubscribe\",\"data\":[\"" << property << "\"],\"transactionid\":\"" << uuid << "\"}";
242 string replystr = s.str();
243 //printf("Reply: %s\n",replystr.c_str());
244 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
246 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
247 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
248 strcpy(new_response,replystr.c_str());
249 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
250 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
253 void WebSocketSinkManager::setValue(string property,string value)
255 AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(property,value);
257 AsyncSetPropertyRequest request;
258 request.property = property;
259 request.value = type;
260 request.completed = [](AsyncPropertyReply* reply)
262 ///TODO: do something here on !reply->success
266 m_engine->setProperty(request);
267 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "AbstractRoutingEngine::setProperty called with arguments:" << property << value << "\n";
271 void WebSocketSinkManager::addSink(libwebsocket* socket, VehicleProperty::Property property,string uuid)
275 //TODO: Dirty hack hardcoded stuff, jsut to make it work.
277 if (property == "running_status_speedometer")
279 tmpstr = VehicleProperty::VehicleSpeed;
281 else if (property == "running_status_engine_speed")
283 tmpstr = VehicleProperty::EngineSpeed;
285 else if (property == "running_status_steering_wheel_angle")
287 tmpstr = VehicleProperty::SteeringWheelAngle;
289 else if (property == "running_status_transmission_gear_status")
291 tmpstr = VehicleProperty::TransmissionShiftPosition;
295 PropertyList foo = VehicleProperty::capabilities();
296 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(property))
302 //Invalid property requested.
307 s << "{\"type\":\"methodReply\",\"name\":\"subscribe\",\"data\":[\"" << property << "\"],\"transactionid\":\"" << uuid << "\"}";
309 string replystr = s.str();
310 //printf("Reply: %s\n",replystr.c_str());
311 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Reply:" << replystr << "\n";
313 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
314 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
315 strcpy(new_response,replystr.c_str());
316 libwebsocket_write(socket, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
317 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
318 WebSocketSink *sink = new WebSocketSink(m_engine,socket,uuid,property,tmpstr);
319 m_sinkMap[property].push_back(sink);
321 extern "C" AbstractSinkManager * create(AbstractRoutingEngine* routingengine, map<string, string> config)
323 sinkManager = new WebSocketSinkManager(routingengine, config);
327 void WebSocketSinkManager::disconnectAll(libwebsocket* socket)
329 std::list<WebSocketSink*> toDeleteList;
331 for (auto i=m_sinkMap.begin(); i != m_sinkMap.end();i++)
333 std::list<WebSocketSink*> *sinks = & (*i).second;
334 for (auto sinkItr = sinks->begin(); sinkItr != sinks->end(); sinkItr++)
336 if ((*sinkItr)->socket() == socket)
338 //This is the sink in question.
339 WebSocketSink* sink = (*sinkItr);
340 if(!ListPlusPlus<WebSocketSink*>(&toDeleteList).contains(sink))
342 toDeleteList.push_back(sink);
345 sinks->erase(sinkItr);
346 sinkItr = sinks->begin();
347 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Sink removed"<<endl;
352 for(auto i=toDeleteList.begin();i!=toDeleteList.end();i++)
357 void WebSocketSinkManager::addPoll(int fd)
359 GIOChannel *chan = g_io_channel_unix_new(fd);
360 guint sourceid = g_io_add_watch(chan,G_IO_IN,(GIOFunc)gioPollingFunc,chan);
361 g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
362 m_ioChannelMap[fd] = chan;
363 m_ioSourceMap[fd] = sourceid;
365 void WebSocketSinkManager::removePoll(int fd)
367 g_io_channel_shutdown(m_ioChannelMap[fd],false,0);
368 //printf("Shutting down IO Channel\n");
369 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Shutting down IO Channel\n";
370 g_source_remove(m_ioSourceMap[fd]); //Since the watch owns the GIOChannel, this should unref it enough to dissapear.
371 //for (map<int,guint>::const_iterator i=m_ioSourceMap.cbegin();i!=m_ioSourceMap.cend();i++)
372 for (map<int,guint>::iterator i=m_ioSourceMap.begin();i!=m_ioSourceMap.end();i++)
376 //printf("Erasing source\n");
377 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing source\n";
378 m_ioSourceMap.erase(i);
382 //for (map<int,GIOChannel*>::const_iterator i=m_ioChannelMap.cbegin();i!=m_ioChannelMap.cend();i++)
383 for (map<int,GIOChannel*>::iterator i=m_ioChannelMap.begin();i!=m_ioChannelMap.end();i++)
387 //printf("Erasing channel\n");
388 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing channel\n";
389 m_ioChannelMap.erase(i);
395 static int websocket_callback(struct libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user,void *in, size_t len)
397 //printf("Switch: %i\n",reason);
402 case LWS_CALLBACK_CLIENT_WRITEABLE:
404 //Connection has been established.
405 //printf("Connection established\n");
408 case LWS_CALLBACK_CLOSED:
410 //Connection is closed, we need to remove all related sinks
411 sinkManager->disconnectAll(wsi);
413 GIOChannel *chan = g_io_channel_unix_new((int)(long)user);
414 g_io_add_watch(chan,G_IO_IN,(GIOFunc)gioPollingFunc,0);
415 g_io_add_watch(chan,G_IO_PRI,(GIOFunc)gioPollingFunc,0);
416 pollfds[count_pollfds].fd = (int)(long)user;
417 pollfds[count_pollfds].events = (int)len;
418 // pollfds[count_pollfds++].revents = 0;*/
421 case LWS_CALLBACK_CLIENT_RECEIVE:
423 //printf("Client writable\n");
426 case LWS_CALLBACK_SERVER_WRITEABLE:
428 //printf("Server writable\n");
432 case LWS_CALLBACK_RECEIVE:
434 //printf("Data Received: %s\n",(char*)in);
435 //The lack of a break; here is intentional.
437 case LWS_CALLBACK_HTTP:
439 //TODO: Verify that ALL requests get sent via LWS_CALLBACK_HTTP, so we can use that instead of LWS_CALLBACK_RECIEVE
440 //TODO: Do we want exceptions, or just to return an invalid json reply? Probably an invalid json reply.
441 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " Requested: " << (char*)in << "\n";
442 GError* error = nullptr;
445 JsonParser* parser = json_parser_new();
446 if (!json_parser_load_from_data(parser,(char*)in,len,&error))
448 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Error loading JSON\n";
452 JsonNode* node = json_parser_get_root(parser);
455 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Error getting root node of json\n";
456 //throw std::runtime_error("Unable to get JSON root object");
460 JsonReader* reader = json_reader_new(node);
461 if(reader == nullptr)
463 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "json_reader is null!\n";
464 //throw std::runtime_error("Unable to create JSON reader");
473 json_reader_read_member(reader,"type");
474 type = json_reader_get_string_value(reader);
475 json_reader_end_member(reader);
478 json_reader_read_member(reader,"name");
479 name = json_reader_get_string_value(reader);
480 json_reader_end_member(reader);
485 json_reader_read_member(reader,"data");
486 if (json_reader_is_array(reader))
488 for(int i=0; i < json_reader_count_elements(reader); i++)
490 json_reader_read_element(reader,i);
491 if (json_reader_is_value(reader))
494 string path = json_reader_get_string_value(reader);
495 data.push_back(path);
500 //Not a raw string value, then it's "property/value" kvp, for "set" requests
501 json_reader_read_member(reader,"property");
502 string keystr = json_reader_get_string_value(reader);
503 key.push_back(keystr);
504 json_reader_end_member(reader);
505 json_reader_read_member(reader,"value");
506 string valuestr = json_reader_get_string_value(reader);
507 value.push_back(valuestr);
508 json_reader_end_member(reader);
510 json_reader_end_element(reader);
515 string path = json_reader_get_string_value(reader);
518 data.push_back(path);
521 json_reader_end_member(reader);
524 json_reader_read_member(reader,"transactionid");
525 if (strcmp("gchararray",g_type_name(json_node_get_value_type(json_reader_get_value(reader)))) == 0)
528 id = json_reader_get_string_value(reader);
534 strstr << json_reader_get_int_value(reader);
537 json_reader_end_member(reader);
539 ///TODO: this will probably explode:
540 //mlc: I agree with Kevron here, it does explode.
541 //if(error) g_error_free(error);
543 g_object_unref(reader);
544 g_object_unref(parser);
547 if (type == "method")
553 //GetProperty is going to be a singleshot sink.
554 //string arg = arguments.front();
555 sinkManager->addSingleShotSink(wsi,data.front(),id);
559 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " \"get\" method called with no data! Transaction ID:" << id << "\n";
562 else if (name == "set")
568 else if (value.size() > 0)
570 if (key.size() != value.size())
572 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "\"set\" method called with an invalid key value pair count\n";
576 list<string>::iterator d = value.begin();
577 for (list<string>::iterator i=key.begin();i!=key.end();i++)
579 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "websocketsinkmanager setting" << (*i) << "to" << (*d) << "\n";
581 sinkManager->setValue((*i),(*d));
589 else if (name == "subscribe")
591 //Websocket wants to subscribe to an event, data.front();
592 for (auto i=data.begin();i!=data.end();i++)
594 sinkManager->addSink(wsi,(*i),id);
597 else if (name == "unsubscribe")
599 //Websocket wants to unsubscribe to an event, data.front();
600 for (auto i=data.begin();i!=data.end();i++)
602 sinkManager->removeSink(wsi,(*i),id);
605 else if (name == "getHistory")
609 std::string property = data[0];
610 std::string startStr = data[1];
611 std::string endStr = data[2];
613 sinkManager->addSingleShotRangedSink(wsi,property,
614 boost::lexical_cast<double,std::string>(startStr),
615 boost::lexical_cast<double,std::string>(endStr), id);
620 //TODO: error, "invalid arguments" should be sent in reply to this.
623 else if (name == "getSupportedEventTypes")
625 //If data.front() dosen't contain a property name, return a list of properties supported.
626 //if it does, then return the event types that particular property supports.
627 string typessupported = "";
628 if (data.size() == 0)
630 //Send what properties we support
631 typessupported = "\"running_status_speedometer\",\"running_status_engine_speed\",\"running_status_steering_wheel_angle\",\"running_status_transmission_gear_status\"";
633 PropertyList foo = sinkManager->getSupportedProperties();
634 PropertyList::const_iterator i=foo.cbegin();
635 while (i != foo.cend())
637 typessupported.append(",\"").append((*i)).append("\"");
643 //Send what events a particular property supports
644 if (data.front()== "running_status_speedometer")
646 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
648 else if (data.front()== "running_status_engine_speed")
650 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
652 else if (data.front() == "running_status_steering_wheel_angle")
654 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
656 else if (data.front() == "running_status_transmission_gear_status")
658 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
662 PropertyList foo = sinkManager->getSupportedProperties();
663 if (ListPlusPlus<VehicleProperty::Property>(&foo).contains(data.front()))
665 //sinkManager->addSingleShotSink(wsi,data.front(),id);
666 typessupported = "\"get\",\"subscribe\",\"unsubscribe\",\"getSupportedEventTypes\"";
672 s << "{\"type\":\"methodReply\",\"name\":\"getSupportedEventTypes\",\"data\":[" << typessupported << "],\"transactionid\":\"" << id << "\"}";
673 string replystr = s.str();
674 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " JSON Reply: " << replystr << "\n";
675 //printf("Reply: %s\n",replystr.c_str());
676 char *new_response = new char[LWS_SEND_BUFFER_PRE_PADDING + strlen(replystr.c_str()) + LWS_SEND_BUFFER_POST_PADDING];
677 new_response+=LWS_SEND_BUFFER_PRE_PADDING;
678 strcpy(new_response,replystr.c_str());
679 libwebsocket_write(wsi, (unsigned char*)new_response, strlen(new_response), LWS_WRITE_TEXT);
680 delete (char*)(new_response-LWS_SEND_BUFFER_PRE_PADDING);
684 DebugOut(0)<<"Unknown method called."<<endl;
689 case LWS_CALLBACK_ADD_POLL_FD:
691 //printf("Adding poll %i\n",sinkManager);
692 //DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Adding poll" << (int)sinkManager << "\n";
693 if (sinkManager != 0)
695 sinkManager->addPoll((int)(long)user);
699 case LWS_CALLBACK_DEL_POLL_FD:
701 sinkManager->removePoll((int)(long)user);
704 case LWS_CALLBACK_SET_MODE_POLL_FD:
709 case LWS_CALLBACK_CLEAR_MODE_POLL_FD:
711 //Don't handle this yet.
716 //printf("Unhandled callback: %i\n",reason);
717 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Unhandled callback:" << reason << "\n";
724 bool gioPollingFunc(GIOChannel *source,GIOCondition condition,gpointer data)
726 if (condition != G_IO_IN)
728 //Don't need to do anything
729 if (condition == G_IO_HUP)
731 //Hang up. Returning false closes out the GIOChannel.
732 //printf("Callback on G_IO_HUP\n");
737 //This is the polling function. If it return false, glib will stop polling this FD.
738 //printf("Polling...%i\n",condition);
740 struct pollfd pollstruct;
741 int newfd = g_io_channel_unix_get_fd(source);
742 pollstruct.fd = newfd;
743 pollstruct.events = condition;
744 pollstruct.revents = condition;
745 libwebsocket_service_fd(context,&pollstruct);