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 "websocketsource.h"
22 #include <boost/assert.hpp>
23 #include <boost/lexical_cast.hpp>
26 #include <listplusplus.h>
28 #include <timestamp.h>
29 #include "uuidhelper.h"
31 #include <QVariantMap>
32 #include <QJsonDocument>
33 #include <QStringList>
37 #include "superptr.hpp"
39 #define __SMALLFILE__ std::string(__FILE__).substr(std::string(__FILE__).rfind("/")+1)
40 libwebsocket_context *context = NULL;
41 WebSocketSource *source;
42 AbstractRoutingEngine *m_re;
44 double oldTimestamp=0;
47 double averageLatency=0;
49 class UniquePropertyCache
52 bool hasProperty(std::string name, std::string source, Zone::Type zone)
54 for(auto i : properties)
57 i->sourceUuid == source &&
66 std::shared_ptr<AbstractPropertyType> append(std::string name, std::string source, Zone::Type zone)
68 for(auto i : properties)
71 i->sourceUuid == source &&
78 auto t = VehicleProperty::getPropertyTypeForPropertyNameValue(name);
82 throw std::runtime_error(name + "name is not a known type");
85 t->sourceUuid = source;
88 properties.emplace_back(t);
90 return property(name, source, zone);
93 std::shared_ptr<AbstractPropertyType> property(std::string name, std::string source, Zone::Type zone)
95 for(auto i : properties)
98 i->sourceUuid == source &&
109 std::vector<std::shared_ptr<AbstractPropertyType>> properties;
112 UniquePropertyCache properties;
114 static int callback_http_only(libwebsocket_context *context,struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason,void *user, void *in, size_t len);
115 static struct libwebsocket_protocols protocols[] = {
130 //Called when a client connects, subscribes, or unsubscribes.
131 void WebSocketSource::checkSubscriptions()
133 while (queuedRequests.size() > 0)
135 VehicleProperty::Property prop = queuedRequests.front();
136 removeOne(&queuedRequests,prop);
137 if (contains(activeRequests,prop))
141 activeRequests.push_back(prop);
145 reply["type"] = "method";
146 reply["name"] = "subscribe";
147 reply["data"] = prop.c_str();
148 reply["transactionid"] = "d293f670-f0b3-11e1-aff1-0800200c9a66";
153 replystr = QJsonDocument::fromVariant(reply).toBinaryData();
156 replystr = QJsonDocument::fromVariant(reply).toJson();
160 lwsWrite(clientsocket, replystr, replystr.length());
163 void WebSocketSource::setConfiguration(map<string, string> config)
165 //printf("WebSocketSource::setConfiguration has been called\n");
168 configuration = config;
170 if(config.find("binaryProtocol") != config.end())
172 doBinary = config["binaryProtocol"] == "true";
175 for (map<string,string>::iterator i=configuration.begin();i!=configuration.end();i++)
177 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming setting for WebSocketSource:" << (*i).first << ":" << (*i).second << "\n";
178 //printf("Incoming setting: %s:%s\n",(*i).first.c_str(),(*i).second.c_str());
179 if ((*i).first == "ip")
183 if ((*i).first == "port")
185 port = boost::lexical_cast<int>((*i).second);
187 if ((*i).first == "ssl")
189 if ((*i).second == "true")
195 m_sslEnabled = false;
199 //printf("Connecting to websocket server at %s port %i\n",ip.c_str(),port);
200 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Connecting to websocket server at" << ip << ":" << port << "\n";
204 DebugOut(5) << "SSL ENABLED" << endl;
208 clientsocket = libwebsocket_client_connect(context, ip.c_str(), port, sslval,"/", "localhost", "websocket",protocols[0].name, -1);
213 PropertyInfo WebSocketSource::getPropertyInfo(VehicleProperty::Property property)
215 return PropertyInfo::invalid();
218 bool gioPollingFunc(GIOChannel *source, GIOCondition condition, gpointer data)
220 //This is the polling function. If it return false, glib will stop polling this FD.
222 oldTimestamp = amb::currentTime();
224 struct pollfd pollstruct;
225 int newfd = g_io_channel_unix_get_fd(source);
226 pollstruct.fd = newfd;
227 pollstruct.events = condition;
228 pollstruct.revents = condition;
229 libwebsocket_service_fd(context,&pollstruct);
230 if (condition & G_IO_HUP)
232 //Hang up. Returning false closes out the GIOChannel.
233 //printf("Callback on G_IO_HUP\n");
236 if (condition & G_IO_IN)
240 DebugOut() << "gioPollingFunc" << condition << endl;
245 static int checkTimeouts(gpointer data)
247 WebSocketSource *src = (WebSocketSource*)data;
248 for (auto i=src->uuidTimeoutMap.begin();i!= src->uuidTimeoutMap.end();i++)
250 if (src->uuidRangedReplyMap.find((*i).first) != src->uuidRangedReplyMap.end())
253 if (amb::currentTime() > (*i).second)
255 //We've reached timeout
256 DebugOut() << "Timeout reached for request ID:" << (*i).first << "\n";
257 src->uuidRangedReplyMap[(*i).first]->success = false;
258 src->uuidRangedReplyMap[(*i).first]->completed(src->uuidRangedReplyMap[(*i).first]);
259 src->uuidRangedReplyMap.erase((*i).first);
260 src->uuidTimeoutMap.erase((*i).first);
263 if (src->uuidTimeoutMap.size() == 0)
271 //No timeout yet, keep waiting.
276 //Reply has already come back, ignore and erase from list.
277 src->uuidTimeoutMap.erase((*i).first);
280 if (src->uuidTimeoutMap.size() == 0)
290 static int callback_http_only(libwebsocket_context *context, struct libwebsocket *wsi,enum libwebsocket_callback_reasons reason, void *user, void *in, size_t len)
292 unsigned char buf[LWS_SEND_BUFFER_PRE_PADDING + 4096 + LWS_SEND_BUFFER_POST_PADDING];
293 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << reason << "callback_http_only" << endl;
296 case LWS_CALLBACK_CLOSED:
297 //fprintf(stderr, "mirror: LWS_CALLBACK_CLOSED\n");
299 //printf("Connection closed!\n");
302 //case LWS_CALLBACK_PROTOCOL_INIT:
303 case LWS_CALLBACK_CLIENT_ESTABLISHED:
305 //This happens when a client initally connects. We need to request the support event types.
306 source->clientConnected = true;
307 source->checkSubscriptions();
308 //printf("Incoming connection!\n");
309 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Incoming connection" << endl;
312 toSend["type"] = "method";
313 toSend["name"] = "getSupportedEventTypes";
314 toSend["transactionid"] = amb::createUuid().c_str();
319 replystr = QJsonDocument::fromVariant(toSend).toBinaryData();
322 replystr = QJsonDocument::fromVariant(toSend).toJson();
326 lwsWrite(wsi, replystr, replystr.length());
330 case LWS_CALLBACK_CLIENT_RECEIVE:
332 QByteArray d((char*)in,len);
334 WebSocketSource * manager = source;
336 if(manager->expectedMessageFrames && manager->partialMessageIndex < manager->expectedMessageFrames)
338 manager->incompleteMessage += d;
339 manager->partialMessageIndex++;
342 else if(manager->expectedMessageFrames && manager->partialMessageIndex == manager->expectedMessageFrames)
344 d = manager->incompleteMessage + d;
345 manager->expectedMessageFrames = 0;
351 doc = QJsonDocument::fromBinaryData(d);
354 doc = QJsonDocument::fromJson(d);
355 DebugOut(7)<<d.data()<<endl;
360 DebugOut(DebugOut::Error)<<"Invalid message"<<endl;
364 QVariantMap call = doc.toVariant().toMap();
366 string type = call["type"].toString().toStdString();
367 string name = call["name"].toString().toStdString();
368 string id = call["transactionid"].toString().toStdString();
370 list<pair<string,string> > pairdata;
372 if(type == "multiframe")
374 manager->expectedMessageFrames = call["frames"].toInt();
375 manager->partialMessageIndex = 1;
376 manager->incompleteMessage = "";
379 else if (type == "valuechanged")
381 QVariantMap data = call["data"].toMap();
383 string value = data["value"].toString().toStdString();
384 double timestamp = data["timestamp"].toDouble();
385 int sequence = data["sequence"].toInt();
386 Zone::Type zone = data["zone"].toInt();
388 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Value changed:" << name << value << endl;
392 auto type = properties.append(name, source->uuid(), zone);
394 type->timestamp = timestamp;
395 type->sequence = sequence;
396 type->fromString(value);
398 m_re->updateProperty(type.get(), source->uuid());
399 double currenttime = amb::currentTime();
401 /** This is now the latency between when something is available to read on the socket, until
402 * a property is about to be updated in AMB. This includes libwebsockets parsing and the
403 * JSON parsing in this section.
406 DebugOut(2)<<"websocket network + parse latency: "<<(currenttime - type->timestamp)*1000<<"ms"<<endl;
407 totalTime += (currenttime - oldTimestamp)*1000;
409 averageLatency = totalTime / numUpdates;
411 DebugOut(2)<<"Average parse latency: "<<averageLatency<<endl;
415 //printf("Exception %s\n",ex.what());
416 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Exception:" << ex.what() << "\n";
419 else if (type == "methodReply")
421 if (name == "getSupportedEventTypes")
424 QVariant data = call["data"];
426 QStringList supported = data.toStringList();
428 DebugOut() << __SMALLFILE__ <<":"<< __LINE__ << "Got getSupportedEventTypes request"<<endl;
431 Q_FOREACH(QString p, supported)
433 props.push_back(p.toStdString());
436 source->setSupported(props);
437 //m_re->updateSupported(m_supportedProperties,PropertyList());
439 else if (name == "getRanged")
441 QVariantList data = call["data"].toList();
443 std::list<AbstractPropertyType*> propertylist;
445 Q_FOREACH(QVariant d, data)
447 QVariantMap obj = d.toMap();
449 std::string name = obj["name"].toString().toStdString();
450 std::string value = obj["value"].toString().toStdString();
451 double timestamp = obj["timestamp"].toDouble();
452 int sequence = obj["sequence"].toInt();
454 AbstractPropertyType* type = VehicleProperty::getPropertyTypeForPropertyNameValue(name,value);
455 type->timestamp = timestamp;
456 type->sequence = sequence;
458 propertylist.push_back(type);
461 if (source->uuidRangedReplyMap.find(id) != source->uuidRangedReplyMap.end())
463 source->uuidRangedReplyMap[id]->values = propertylist;
464 source->uuidRangedReplyMap[id]->success = true;
465 source->uuidRangedReplyMap[id]->completed(source->uuidRangedReplyMap[id]);
466 source->uuidRangedReplyMap.erase(id);
470 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";
473 else if (name == "get")
476 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "Got \"GET\" event:" << pairdata.size()<<endl;
477 if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end())
479 QVariantMap obj = call["data"].toMap();
481 std::string property = obj["property"].toString().toStdString();
482 std::string value = obj["value"].toString().toStdString();
483 double timestamp = obj["timestamp"].toDouble();
484 int sequence = obj["sequence"].toInt();
485 Zone::Type zone = obj["zone"].toInt();
487 AbstractPropertyType* v = VehicleProperty::getPropertyTypeForPropertyNameValue(property, value);
488 v->timestamp = timestamp;
489 v->sequence = sequence;
492 if (source->uuidReplyMap.find(id) != source->uuidReplyMap.end() && source->uuidReplyMap[id]->error != AsyncPropertyReply::Timeout)
494 source->uuidReplyMap[id]->value = v;
495 source->uuidReplyMap[id]->success = true;
496 source->uuidReplyMap[id]->completed(source->uuidReplyMap[id]);
497 source->uuidReplyMap.erase(id);
502 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";
509 DebugOut() << __SMALLFILE__ << ":" << __LINE__ << "GET Method Reply INVALID! Multiple properties detected, only single are supported!!!" << "\n";
512 //data will contain a property/value map.
520 case LWS_CALLBACK_CLIENT_CONFIRM_EXTENSION_SUPPORTED:
522 //printf("Requested extension: %s\n",(char*)in);
526 case LWS_CALLBACK_ADD_POLL_FD:
528 DebugOut(5) << __SMALLFILE__ << ":" << __LINE__ << "Adding poll for websocket IO channel" << endl;
529 //Add a FD to the poll list.
530 GIOChannel *chan = g_io_channel_unix_new(libwebsocket_get_socket_fd(wsi));
532 /// TODO: I changed this to be more consistent with the websocket sink end. it may not be correct. TEST
534 g_io_add_watch(chan,GIOCondition(G_IO_IN | G_IO_PRI | G_IO_ERR | G_IO_HUP),(GIOFunc)gioPollingFunc,0);
535 g_io_channel_set_close_on_unref(chan,true);
536 g_io_channel_unref(chan); //Pass ownership of the GIOChannel to the watch.
543 void WebSocketSource::setSupported(PropertyList list)
545 DebugOut() <<__SMALLFILE__ << ":" << __LINE__ <<" "<< __FUNCTION__ <<endl;
546 m_supportedProperties = list;
547 m_re->updateSupported(list,PropertyList(),this);
550 WebSocketSource::WebSocketSource(AbstractRoutingEngine *re, map<string, string> config) : AbstractSource(re, config), partialMessageIndex(0),expectedMessageFrames(0)
552 m_sslEnabled = false;
553 clientConnected = false;
556 struct lws_context_creation_info info;
557 memset(&info, 0, sizeof info);
558 info.protocols = protocols;
559 info.extensions = nullptr;
561 if(config.find("useExtensions") != config.end() && config["useExtensions"] == "true")
563 info.extensions = libwebsocket_get_internal_extensions();
568 info.port = CONTEXT_PORT_NO_LISTEN;
571 context = libwebsocket_create_context(&info);
573 setConfiguration(config);
575 //printf("websocketsource loaded!!!\n");
576 g_timeout_add(1000,checkTimeouts,this); //Do this once per second, check for functions that have timed out and reply with success = false;
579 PropertyList WebSocketSource::supported()
581 return m_supportedProperties;
584 int WebSocketSource::supportedOperations()
586 /// TODO: need to do this correctly based on what the host supports.
587 return Get | Set | GetRanged;
590 const string WebSocketSource::uuid()
592 return "d293f670-f0b3-11e1-aff1-0800200c9a66";
595 void WebSocketSource::subscribeToPropertyChanges(VehicleProperty::Property property)
597 //printf("Subscribed to property: %s\n",property.c_str());
598 queuedRequests.push_back(property);
601 checkSubscriptions();
606 void WebSocketSource::unsubscribeToPropertyChanges(VehicleProperty::Property property)
608 removeRequests.push_back(property);
611 checkSubscriptions();
616 void WebSocketSource::getPropertyAsync(AsyncPropertyReply *reply)
618 std::string uuid = amb::createUuid();
619 uuidReplyMap[uuid] = reply;
620 uuidTimeoutMap[uuid] = amb::currentTime() + 10.0; ///TODO: 10 second timeout, make this configurable?
623 data["property"] = reply->property.c_str();
624 data["zone"] = reply->zoneFilter;
626 QVariantMap replyvar;
627 replyvar["type"] = "method";
628 replyvar["name"] = "get";
629 replyvar["data"] = data;
630 replyvar["transactionid"] = uuid.c_str();
635 replystr = QJsonDocument::fromVariant(replyvar).toBinaryData();
638 replystr = QJsonDocument::fromVariant(replyvar).toJson();
642 lwsWrite(clientsocket, replystr, replystr.length());
645 void WebSocketSource::getRangePropertyAsync(AsyncRangePropertyReply *reply)
647 std::string uuid = amb::createUuid();
648 uuidRangedReplyMap[uuid] = reply;
649 uuidTimeoutMap[uuid] = amb::currentTime() + 60; ///TODO: 60 second timeout, make this configurable?
652 s << "{\"type\":\"method\",\"name\":\"getRanged\",\"data\": {";
654 QVariantMap replyvar;
655 replyvar["type"] = "method";
656 replyvar["name"] = "getRanged";
657 replyvar["transactionid"] = uuid.c_str();
658 replyvar["timeBegin"] = reply->timeBegin;
659 replyvar["timeEnd"] = reply->timeEnd;
660 replyvar["sequenceBegin"] = reply->sequenceBegin;
661 replyvar["sequenceEnd"] = reply->sequenceEnd;
664 QStringList properties;
666 for (auto itr = reply->properties.begin(); itr != reply->properties.end(); itr++)
668 VehicleProperty::Property p = *itr;
669 properties.append(p.c_str());
672 replyvar["data"] = properties;
677 replystr = QJsonDocument::fromVariant(replyvar).toBinaryData();
680 replystr = QJsonDocument::fromVariant(replyvar).toJson();
684 lwsWrite(clientsocket, replystr, replystr.length());
687 AsyncPropertyReply * WebSocketSource::setProperty( AsyncSetPropertyRequest request )
689 AsyncPropertyReply* reply = new AsyncPropertyReply(request);
692 data["property"] = request.property.c_str();
693 data["value"] = request.value->toString().c_str();
694 data["zone"] = request.zoneFilter;
697 QVariantMap replyvar;
698 replyvar["type"] = "method";
699 replyvar["name"] = "set";
700 replyvar["data"] = data;
701 replyvar["transactionid"] = amb::createUuid().c_str();
706 replystr = QJsonDocument::fromVariant(replyvar).toBinaryData();
709 replystr = QJsonDocument::fromVariant(replyvar).toJson();
713 lwsWrite(clientsocket, replystr, replystr.length());
715 ///TODO: we should actually wait for a response before we simply complete the call
716 reply->success = true;
717 reply->completed(reply);
721 extern "C" AbstractSource * create(AbstractRoutingEngine* routingengine, map<string, string> config)
723 return new WebSocketSource(routingengine, config);