Add notifications through websockets

src/lib/common/limits.h

@@ -115,6 +115,13 @@
 #define HTTP_HEADER_HOST_MAX_LENGTH         256
 
 
+/* ****************************************************************************
+*
+* Subscription id maximun length - 
+*/
+#define MAX_LENGTH_SUBID    24
+
+
 /* ****************************************************************************
 *
 * Default timeout - 5000 milliseconds

src/lib/common/string.cpp

@@ -35,6 +35,7 @@
 #include "common/wsStrip.h"
 #include "alarmMgr/alarmMgr.h"
 
+#include "orion_websocket/constants.h"
 
 
 /* ****************************************************************************
@@ -242,6 +243,10 @@ bool parseUrl(const std::string& url, std::string& host, int& port, std::string&
     return false;
   }
 
+ if (url == WSConstants::Scheme)
+ {
+    return true;
+ }
 
   /* http://some.host.com/my/path
    *      ^^             ^  ^

src/lib/ngsiNotify/Notifier.cpp

@@ -35,10 +35,7 @@
 #include "ngsiNotify/senderThread.h"
 #include "ngsiNotify/Notifier.h"
 
-
-// FIME P11 #1669: move this to WS library. @fortizc please take care of this
-#define WS_SCHEME         "ws://"
-#define WS_SCHEME_LENGTH  5
+#include "orion_websocket/constants.h"
 
 
 /* ****************************************************************************
@@ -99,10 +96,10 @@ void Notifier::sendNotifyContextRequest(NotifyContextRequest* ncr, const std::st
     std::string  uriPath;
     std::string  protocol;
 
-    if ((url.length() == WS_SCHEME_LENGTH) && (url.find(WS_SCHEME) == 0))
+    if ((url.length() == WSConstants::Scheme.size()) && (url.find(WSConstants::Scheme) == 0))
     {
       // In this case host, port and uriPath are not needed, as the WS library has all the connection information related with the WS
-      protocol = WS_SCHEME;
+      protocol = WSConstants::Scheme;
     }
     else
     {

src/lib/ngsiNotify/senderThread.cpp

@@ -29,18 +29,8 @@
 #include "rest/httpRequestSend.h"
 #include "ngsiNotify/senderThread.h"
 
-
-// FIXME P11 #1669: remove this stub when the actual sendNotifyContextRequestWs() gets developed a the end. @fortizc please take care of this
-// in the ws library
-int sendNotifyContextRequestWs(const std::string& subId, const std::map<std::string, std::string>& headers, const std::string& data)
-{
-  return 0;
-}
-
-
-// FIME P11 #1669: move this to WS library. @fortizc please take care of this
-#define WS_SCHEME         "ws://"
-#define WS_SCHEME_LENGTH  5
+#include "orion_websocket/constants.h"
+#include "orion_websocket/wsNotify.h"
 
 
 /* ****************************************************************************
@@ -73,7 +63,7 @@ void* startSenderThread(void* p)
       std::string  out;
       int          r;
 
-      if (params->protocol == WS_SCHEME)
+      if (params->protocol == WSConstants::Scheme)
       {
         std::map<std::string, std::string> headers;
         headers.insert(std::make_pair("Fiware-Service",     params->tenant));

src/lib/orion_websocket/CMakeLists.txt

@@ -22,14 +22,16 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.6)
 
 SET (SOURCES
     ws.cpp
-    connection_manager.cpp
     parser.cpp
+    constants.cpp
+    wsNotify.cpp
 )
 
 SET (HEADERS
     ws.h
-    connection_manager.h
     parser.h
+    constants.h
+    wsNotify.h
 )
 
 

src/lib/orion_websocket/constants.cpp

@@ -0,0 +1,35 @@
+/*
+*
+* Copyright 2016 Telefonica Investigacion y Desarrollo, S.A.U
+*
+* This file is part of Orion Context Broker.
+*
+* Orion Context Broker is free software: you can redistribute it and/or
+* modify it under the terms of the GNU Affero General Public License as
+* published by the Free Software Foundation, either version 3 of the
+* License, or (at your option) any later version.
+*
+* Orion Context Broker is distributed in the hope that it will be useful,
+* but WITHOUT ANY WARRANTY; without even the implied warranty of
+* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU Affero
+* General Public License for more details.
+*
+* You should have received a copy of the GNU Affero General Public License
+* along with Orion Context Broker. If not, see http://www.gnu.org/licenses/.
+*
+* For those usages not covered by this license please contact with
+* iot_support at tid dot es
+*
+* Author: Felipe Ortiz
+*/
+
+
+#include "constants.h"
+
+const int WSConstants::Port = 9010;
+const int WSConstants::Pooling = 50;
+const size_t WSConstants::DataSize = 128;
+// Ok, this is can be a macro, but I prefer
+// to be consistent
+const std::string WSConstants::ProtocolName = "ngsiv2-json";
+const std::string WSConstants::Scheme = "ws://";

src/lib/orion_websocket/connection_manager.h → src/lib/orion_websocket/constants.h

@@ -1,6 +1,6 @@
 /*
 *
-* Copyright 2015 Telefonica Investigacion y Desarrollo, S.A.U
+* Copyright 2016 Telefonica Investigacion y Desarrollo, S.A.U
 *
 * This file is part of Orion Context Broker.
 *
@@ -24,13 +24,20 @@
 */
 
 
-#ifndef WS_CONNECTION_MANAGER_H
-#define WS_CONNECTION_MANAGER_H
+#ifndef WS_CONSTANTS_H
+#define WS_CONSTANTS_H
 
-class ConnectionInfo;
+#include <string>
 
-int connection_manager_get_cid();
-ConnectionInfo *connection_manager_get(int cid, const char *msg);
-void connection_manager_remove(int cid);
+
+class WSConstants
+{
+public:
+  static const int Port;
+  static const int Pooling;
+  static const size_t DataSize;
+  static const std::string ProtocolName;
+  static const std::string Scheme;
+};
 
 #endif

src/lib/orion_websocket/parser.cpp

@@ -103,10 +103,31 @@ void ws_parser_parse
   }
 }
 
+static void addHeaders
+(
+ const std::string& name,
+ const std::string& value,
+ rapidjson::Writer<rapidjson::StringBuffer>& writer
+)
+{
+  writer.Key(name.c_str());
+
+  if (value.empty())
+  {
+    writer.String("");
+  }
+  else
+  {
+    writer.String(value.c_str());
+  }
+}
+
 const char *ws_parser_message
 (
  const std::string&  msg,
  const HttpHeaders&  head,
+ const std::vector<std::string> headName,
+ const std::vector<std::string> headValue,
  int                 statusCode
 )
 {
@@ -119,18 +140,15 @@ const char *ws_parser_message
   std::map<std::string, std::string *>::const_iterator it = head.headerMap.begin();
   while (it != head.headerMap.end())
   {
-    writer.Key(it->first.c_str());
-
-    if (it->second->empty())
-    {
-        writer.String("");
-    }
-    else
-    {
-        writer.String(it->second->c_str());
-    }
+    addHeaders(it->first, *(it->second), writer);
     ++it;
   }
+
+  for (unsigned i = 0; i < headName.size(); ++i)
+  {
+    addHeaders(headName[i], headValue[i], writer);
+  }
+
   writer.EndObject();
 
   if (statusCode < 100 || statusCode > 599) // Code is not a valid HTTP status code
@@ -146,3 +164,34 @@ const char *ws_parser_message
 
   return json;
 }
+
+const char *ws_parser_notify
+(
+ const std::string& subId,
+ const std::map<std::string, std::string>& headers,
+ const std::string &data
+)
+{
+  const char* tmpl = "{\"subscriptionId\": \"%s\", \"headers\": %s, \"data\": %s}";
+
+
+  rapidjson::StringBuffer buff;
+  rapidjson::Writer<rapidjson::StringBuffer> writer(buff);
+  writer.StartObject();
+  std::map<std::string, std::string>::const_iterator it = headers.begin();
+  while (it != headers.end())
+  {
+    writer.Key(it->first.c_str());
+    writer.String(it->second.c_str());
+    ++it;
+  }
+  writer.EndObject();
+
+  const char* strHeaders = buff.GetString();
+
+  size_t size = subId.size() + data.size() + strlen(strHeaders) + strlen(tmpl) - 6 + 1;
+  char *json = (char *) malloc(size);
+  sprintf(json, tmpl, subId.c_str(), strHeaders, data.c_str());
+
+  return json;
+}

src/lib/orion_websocket/parser.h

@@ -27,6 +27,7 @@
 #define WS_PARSER_H
 
 #include <vector>
+#include <map>
 #include <string>
 
 class HttpHeaders;
@@ -46,7 +47,16 @@ const char *ws_parser_message
 (
  const std::string&  msg,
  const HttpHeaders&  head,
+ const std::vector<std::string> headName,
+ const std::vector<std::string> headValue,
  int                 statusCode
 );
 
+const char *ws_parser_notify
+(
+ const std::string& subId,
+ const std::map<std::string, std::string>& headers,
+ const std::string& data
+);
+
 #endif