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
19 #include "websocketsinkmanager.h"
20 #include "websocketsink.h"
24 #include <json/json.h>
25 #include <json/json_object.h>
26 #include <json/json_tokener.h>
27 #include <listplusplus.h>
30 #include <QVariantMap>
31 #include <QJsonDocument>
32 #include <QStringList>
35 #define __SMALLFILE__ std::string(__FILE__).substr(std::string(__FILE__).rfind("/")+1)
37 //Global variables, these will be moved into the class
38 struct pollfd pollfds[100];
39 int count_pollfds = 0;
40 libwebsocket_context *context;
41 WebSocketSinkManager *sinkManager;
42 static int websocket_callback(struct libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user,void *in, size_t len);
43 bool gioPollingFunc(GIOChannel *source,GIOCondition condition,gpointer data);
45 WebSocketSinkManager::WebSocketSinkManager(AbstractRoutingEngine* engine, map<string, string> config)
46 :routingEngine(engine), configuration(config), partialMessageIndex(0), expectedMessageFrames(0)
48 if(config.find("binaryProtocol") != config.end())
50 doBinary = config["binaryProtocol"] == "true";
53 //Create a listening socket on port 23000 on localhost.
58 void WebSocketSinkManager::init()
60 //Protocol list for libwebsockets.
61 protocollist[0] = { "http-only", websocket_callback, 0 };
62 protocollist[1] = { NULL, NULL, 0 };
65 setConfiguration(configuration);
68 PropertyList WebSocketSinkManager::getSupportedProperties()
70 return m_engine->supported();
73 void WebSocketSinkManager::setConfiguration(map<string, string> config)
75 // //Config has been passed, let's start stuff up.
76 configuration = config;
77 struct lws_context_creation_info info;
78 memset(&info, 0, sizeof info);
82 std::string interface = "lo";
83 std::string ssl_cert_path;
84 std::string ssl_key_path;
87 info.extensions = nullptr;
90 for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
92 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
93 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting:" << (*i).first << ":" << (*i).second << "\n";
94 if ((*i).first == "interface")
96 interface = (*i).second;
98 if ((*i).first == "port")
100 port = boost::lexical_cast<int>((*i).second);
102 if ((*i).first == "cert")
104 ssl_cert_path = (*i).second;
106 if ((*i).first == "key")
108 ssl_key_path = (*i).second;
110 if ((*i).first == "ssl")
112 if ((*i).second == "true")
121 if ((*i).first == "useExtensions")
124 if((*i).second == "true")
126 info.extensions = libwebsocket_get_internal_extensions();
128 else info.extensions = nullptr;
132 info.iface = interface.c_str();
133 info.protocols = protocollist;
136 info.options = options;
141 info.ssl_cert_filepath = ssl_cert_path.c_str();
142 info.ssl_private_key_filepath = ssl_key_path.c_str();
144 context = libwebsocket_create_context(&info);
148 void WebSocketSinkManager::addSingleShotSink(libwebsocket* socket, VehicleProperty::Property property, Zone::Type zone, string id)
150 AsyncPropertyRequest request;
152 request.property = property;
153 request.zoneFilter = zone;
154 request.completed = [socket,id,property](AsyncPropertyReply* reply)
156 DebugOut()<<"Got property: "<<reply->property.c_str()<<endl;
157 if(!reply->success || !reply->value)
159 DebugOut()<<"Property value is null"<<endl;
165 data["property"] = property.c_str();
166 data["zone"] = reply->value->zone;
167 data["value"] = reply->value->toString().c_str();
168 data["timestamp"] = reply->value->timestamp;
169 data["sequence"] = reply->value->sequence;
171 QVariantMap replyvar;
173 replyvar["type"]="methodReply";
174 replyvar["name"]="get";
175 replyvar["data"]= data;
176 replyvar["transactionid"]=id.c_str();
178 lwsWriteVariant(socket, replyvar);
183 AsyncPropertyReply* reply = routingEngine->getPropertyAsync(request);
186 void WebSocketSinkManager::addSingleShotRangedSink(libwebsocket* socket, PropertyList properties, double start, double end, double seqstart,double seqend, string id)
188 AsyncRangePropertyRequest rangedRequest;
190 rangedRequest.timeBegin = start;
191 rangedRequest.timeEnd = end;
192 rangedRequest.sequenceBegin = seqstart;
193 rangedRequest.sequenceEnd = seqend;
195 rangedRequest.completed = [socket,id](AsyncRangePropertyReply* reply)
197 QVariantMap replyvar;
200 std::list<AbstractPropertyType*> values = reply->values;
201 for(auto itr = values.begin(); itr != values.end(); itr++)
204 obj["value"]= (*itr)->toString().c_str();
205 obj["timestamp"] = (*itr)->timestamp;
206 obj["sequence"] = (*itr)->sequence;
211 replyvar["type"]="methodReply";
212 replyvar["name"]="getRanged";
213 replyvar["data"]=list;
214 replyvar["transactionid"]=id.c_str();
216 lwsWriteVariant(socket, replyvar);
221 routingEngine->getRangePropertyAsync(rangedRequest);
224 void WebSocketSinkManager::removeSink(libwebsocket* socket,VehicleProperty::Property property, string uuid)
226 if (m_sinkMap.find(property) != m_sinkMap.end())
228 list<WebSocketSink*> sinks = m_sinkMap[property];
232 m_sinkMap[property].remove(i);
237 reply["type"]="methodReply";
238 reply["name"]="unsubscribe";
239 reply["property"]=property.c_str();
240 reply["transactionid"]= uuid.c_str();
242 lwsWriteVariant(socket, reply);
245 void WebSocketSinkManager::setValue(libwebsocket* socket,VehicleProperty::Property property,string value,Zone::Type zone,string uuid)
247 AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(property,value);
249 AsyncSetPropertyRequest request;
250 request.property = property;
251 request.value = type;
252 request.zoneFilter = zone;
253 request.completed = [&](AsyncPropertyReply* reply)
256 data["property"] = property.c_str();
258 data["source"] = reply->value->sourceUuid.c_str();
259 data["success"] = reply->success;
261 QVariantMap replyvar;
262 replyvar["type"]="methodReply";
263 replyvar["name"]="set";
264 replyvar["data"]= data;
265 replyvar["transactionid"]=uuid.c_str();
267 lwsWriteVariant(socket, replyvar);
272 m_engine->setProperty(request);
273 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "AbstractRoutingEngine::setProperty called with arguments:" << property << value << "\n";
277 void WebSocketSinkManager::addSink(libwebsocket* socket, VehicleProperty::Property property, string uuid)
281 reply["type"] = "methodReply";
282 reply["name"] = "subscribe";
283 reply["property"] = property.c_str();
284 reply["transactionid"] = uuid.c_str();
286 lwsWriteVariant(socket, reply);
288 WebSocketSink *sink = new WebSocketSink(m_engine, socket, uuid, property, property);
289 m_sinkMap[property].push_back(sink);
291 extern "C" void create(AbstractRoutingEngine* routingengine, map<string, string> config)
293 sinkManager = new WebSocketSinkManager(routingengine, config);
296 void WebSocketSinkManager::disconnectAll(libwebsocket* socket)
298 std::list<WebSocketSink*> toDeleteList;
300 for (auto i=m_sinkMap.begin(); i != m_sinkMap.end();i++)
302 std::list<WebSocketSink*> *sinks = & (*i).second;
303 for (auto sinkItr = sinks->begin(); sinkItr != sinks->end(); sinkItr++)
305 if ((*sinkItr)->socket() == socket)
307 //This is the sink in question.
308 WebSocketSink* sink = (*sinkItr);
309 if(!contains(toDeleteList, sink))
311 toDeleteList.push_back(sink);
314 sinks->erase(sinkItr);
315 sinkItr = sinks->begin();
316 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Sink removed"<<endl;
321 for(auto i=toDeleteList.begin();i!=toDeleteList.end();i++)
326 void WebSocketSinkManager::addPoll(int fd)
328 GIOChannel *chan = g_io_channel_unix_new(fd);
329 guint sourceid = g_io_add_watch(chan, GIOCondition(G_IO_IN | G_IO_HUP | G_IO_ERR),(GIOFunc)gioPollingFunc,chan);
330 g_io_channel_set_close_on_unref(chan,true);
331 g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
332 m_ioChannelMap[fd] = chan;
333 m_ioSourceMap[fd] = sourceid;
335 void WebSocketSinkManager::removePoll(int fd)
337 g_io_channel_shutdown(m_ioChannelMap[fd],false,0);
338 //printf("Shutting down IO Channel\n");
339 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Shutting down IO Channel\n";
340 g_source_remove(m_ioSourceMap[fd]); //Since the watch owns the GIOChannel, this should unref it enough to dissapear.
342 //for (map<int,guint>::const_iterator i=m_ioSourceMap.cbegin();i!=m_ioSourceMap.cend();i++)
343 for (map<int,guint>::iterator i=m_ioSourceMap.begin();i!=m_ioSourceMap.end();i++)
347 //printf("Erasing source\n");
348 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing source\n";
349 m_ioSourceMap.erase(i);
351 if (m_ioSourceMap.size() == 0)
357 //for (map<int,GIOChannel*>::const_iterator i=m_ioChannelMap.cbegin();i!=m_ioChannelMap.cend();i++)
358 for (map<int,GIOChannel*>::iterator i=m_ioChannelMap.begin();i!=m_ioChannelMap.end();i++)
362 //printf("Erasing channel\n");
363 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Erasing channel\n";
364 m_ioChannelMap.erase(i);
366 if (m_ioChannelMap.size() == 0)
374 static int websocket_callback(struct libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user,void *in, size_t len)
376 //printf("Switch: %i\n",reason);
377 DebugOut(5) << __SMALLFILE__ << ":" << __LINE__ << "websocket_callback:" << reason << endl;
382 case LWS_CALLBACK_CLIENT_WRITEABLE:
386 case LWS_CALLBACK_CLOSED:
388 sinkManager->disconnectAll(wsi);
391 case LWS_CALLBACK_CLIENT_RECEIVE:
395 case LWS_CALLBACK_SERVER_WRITEABLE:
400 case LWS_CALLBACK_RECEIVE:
404 case LWS_CALLBACK_HTTP:
406 //TODO: Verify that ALL requests get sent via LWS_CALLBACK_HTTP, so we can use that instead of LWS_CALLBACK_RECIEVE
407 //TODO: Do we want exceptions, or just to return an invalid json reply? Probably an invalid json reply.
408 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << " Requested: " << (char*)in << "\n";
410 QByteArray d((char*)in,len);
412 WebSocketSinkManager * manager = sinkManager;
414 if(manager->expectedMessageFrames && manager->partialMessageIndex < manager->expectedMessageFrames)
416 manager->incompleteMessage += d;
417 manager->partialMessageIndex++;
420 else if(manager->expectedMessageFrames && manager->partialMessageIndex == manager->expectedMessageFrames)
422 d = manager->incompleteMessage + d;
423 manager->expectedMessageFrames = 0;
428 doc = QJsonDocument::fromBinaryData(d);
430 doc = QJsonDocument::fromJson(d);
434 DebugOut(DebugOut::Error)<<"Invalid message"<<endl;
438 QVariantMap call = doc.toVariant().toMap();
440 string type = call["type"].toString().toStdString();
441 string name = call["name"].toString().toStdString();
442 string id = call["transactionid"].toString().toStdString();
444 if (type == "multiframe")
447 manager->expectedMessageFrames = call["frames"].toInt();
448 manager->partialMessageIndex = 1;
449 manager->incompleteMessage = "";
451 else if (type == "method")
453 if(name == "getRanged")
455 QVariantMap data = call["data"].toMap();
457 PropertyList propertyList;
459 propertyList.push_back(data["property"].toString().toStdString());
461 double timeBegin = data["timeBegin"].toDouble();
462 double timeEnd = data["timeEnd"].toDouble();
463 double sequenceBegin = data["sequenceBegin"].toInt();
464 double sequenceEnd = data["sequenceEnd"].toInt();
466 if ((timeBegin < 0 && timeEnd > 0) || (timeBegin > 0 && timeEnd < 0))
468 DebugOut(DebugOut::Warning)<<"Invalid time begin/end pair"<<endl;
470 else if ((sequenceBegin < 0 && sequenceEnd > 0) || (sequenceBegin > 0 && sequenceEnd < 0))
472 DebugOut(DebugOut::Warning)<<"Invalid sequence begin/end pair"<<endl;
476 sinkManager->addSingleShotRangedSink(wsi,propertyList,timeBegin,timeEnd,sequenceBegin,sequenceEnd,id);
479 else if (name == "get")
481 QVariantMap data = call["data"].toMap();
482 Zone::Type zone = Zone::None;
483 if(data.contains("zone"))
485 zone = data["zone"].toInt();
487 sinkManager->addSingleShotSink(wsi,data["property"].toString().toStdString(),zone,id);
490 else if (name == "set")
492 QVariantMap data = call["data"].toMap();
493 Zone::Type zone(Zone::None);
494 if(data.contains("zone"))
496 zone = data["zone"].toInt();
498 sinkManager->setValue(wsi,data["property"].toString().toStdString(), data["value"].toString().toStdString(), zone, id);
500 else if (name == "subscribe")
502 std::string property = call["property"].toString().toStdString();
503 sinkManager->addSink(wsi, property, id);
505 else if (name == "unsubscribe")
507 sinkManager->removeSink(wsi, call["property"].toString().toStdString(), id);
509 else if (name == "getSupportedEventTypes" || name == "getSupported")
514 PropertyList supported = sinkManager->getSupportedProperties();
515 DebugOut() << "we support " << supported.size() << " properties" << endl;
516 for(VehicleProperty::Property i : supported)
518 std::vector<std::string> sources = sinkManager->router()->sourcesForProperty(i);
519 for(auto source : sources)
521 PropertyInfo info = sinkManager->router()->getPropertyInfo(i, source);
523 for(auto zone : info.zones())
525 auto property = amb::make_unique(VehicleProperty::getPropertyTypeForPropertyNameValue(i));
527 std::string signature = property->signature();
528 const std::string basicType = amb::BasicTypes::fromSignature(signature);
532 map["name"] = i.c_str();
533 map["type"] = basicType.c_str();
534 map["source"] = source.c_str();
541 reply["type"] = "methodReply";
542 reply["name"] = "getSupported";
543 reply["transactionid"] = id.c_str();
544 reply["data"] = list;
546 lwsWriteVariant(wsi, reply);
550 DebugOut(0)<<"Unknown method called."<<endl;
555 case LWS_CALLBACK_ADD_POLL_FD:
557 //printf("Adding poll %i\n",sinkManager);
558 DebugOut(5) << __SMALLFILE__ <<":"<< __LINE__ << "Adding poll" << endl;
559 if (sinkManager != 0)
561 //sinkManager->addPoll((int)(long)user);
562 sinkManager->addPoll(libwebsocket_get_socket_fd(wsi));
566 DebugOut(5) << "Error, invalid sink manager!!" << endl;
570 case LWS_CALLBACK_DEL_POLL_FD:
572 sinkManager->removePoll(libwebsocket_get_socket_fd(wsi));
575 case LWS_CALLBACK_SET_MODE_POLL_FD:
580 case LWS_CALLBACK_CLEAR_MODE_POLL_FD:
582 //Don't handle this yet.
587 //printf("Unhandled callback: %i\n",reason);
588 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Unhandled callback:" << reason << "\n";
595 bool gioPollingFunc(GIOChannel *source, GIOCondition condition, gpointer data)
597 DebugOut(5) << "Polling..." << condition << endl;
599 if(condition & G_IO_ERR)
601 DebugOut(0)<< __SMALLFILE__ <<":"<< __LINE__ <<" websocketsink polling error."<<endl;
604 if (condition & G_IO_HUP)
606 //Hang up. Returning false closes out the GIOChannel.
607 //printf("Callback on G_IO_HUP\n");
608 DebugOut(0)<<"socket hangup event..."<<endl;
612 //This is the polling function. If it return false, glib will stop polling this FD.
613 //printf("Polling...%i\n",condition);
616 struct pollfd pollstruct;
617 int newfd = g_io_channel_unix_get_fd(source);
618 pollstruct.fd = newfd;
619 pollstruct.events = condition;
620 pollstruct.revents = condition;
621 libwebsocket_service_fd(context,&pollstruct);