HTTP Server superpowers for the Collect Agent. Includes HTTP error handling and thread pools.

CollectAgent/Makefile

 include ../config.mk
 
-CXXFLAGS =	-O2 -g --std=c++11 -Wall -Wno-unused-local-typedefs -Wno-deprecated-declarations -Wno-unknown-warning-option -fmessage-length=0 -I$(DCDBDEPLOYPATH)/include/ -I$(DCDBBASEPATH)/include/ -DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG
+CXXFLAGS =	-O2 -g --std=c++11 -Wall -Wno-unused-local-typedefs -Wno-deprecated-declarations -Wno-unknown-warning-option -fmessage-length=0 -I$(DCDBDEPLOYPATH)/include/ -I$(DCDBBASEPATH)/include/ -DBOOST_DATE_TIME_POSIX_TIME_STD_CONFIG -I$(DCDBDEPSPATH)/cpp-netlib-0.12.0-final/deps/asio/asio/include -I$(DCDBDEPSPATH)/cpp-netlib-0.12.0-final -I$(DCDBDEPSPATH)/cpp-netlib-0.12.0-final/deps/cxxopts/src -DASIO_HEADER_ONLY -DBOOST_TEST_DYN_LINK
+
 OBJS =		collectagent.o \
 		sensorcache.o \
 		simplemqttserver.o \
 		simplemqttserverthread.o \
 		simplemqttservermessage.o
-LIBS =		-L$(DCDBDEPLOYPATH)/lib/ -ldcdb -lpthread -lcassandra -luv -lboost_system -lboost_random -lboost_thread -lboost_date_time -lboost_regex -lssl -lcrypto
+LIBS =		-L$(DCDBDEPLOYPATH)/lib/ -ldcdb -pthread -lcassandra -luv -lboost_system -lboost_random -lboost_thread -lboost_date_time -lboost_regex -lcppnetlib-server-parsers
 TARGET =	collectagent
 
 .PHONY : clean install

CollectAgent/collectagent.cpp

@@ -24,6 +24,9 @@
 // Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
 //================================================================================
 
+#include <boost/network/protocol/http/server.hpp>
+#include <boost/network/utils/thread_pool.hpp>
+#include <iostream>
 #include <cstdlib>
 #include <signal.h>
 #include <unistd.h>
@@ -66,6 +69,61 @@ void abrtHandler(int sig)
   abrt(EXIT_FAILURE, SIGNAL);
 }
 
+namespace http = boost::network::http;
+
+/*<< Defines the server. >>*/
+struct http_handler;
+typedef http::server<http_handler> server;
+
+/*<< Defines the request handler.  It's a class that defines two
+     functions, `operator()` and `log()` >>*/
+struct http_handler {
+	
+	/*<< This is the function that handles the incoming request. >>*/
+    void operator()(server::request const &request, server::connection_ptr connection) {
+    server::string_type ip = source(request);
+    
+    DCDB::SensorId sid;
+    sid.mqttTopicConvert(request.destination);
+    
+    std::ostringstream data;
+    
+    static server::response_header headers[] = {{"Connection", "close"},
+												{"Content-Type", "text/plain"}};
+    //try getting the latest value 
+    try {
+		
+		uint64_t val = mySensorCache.getSensor(sid);
+		
+		data << val;
+		//data << "Sid : " << sid.toString() << ", Value: " << val << "." << std::endl;
+    
+		connection->set_status(server::connection::ok);
+		connection->set_headers(boost::make_iterator_range(headers, headers + 2));
+		connection->write(data.str());
+	
+	}
+    
+    catch(const std::invalid_argument& e) {
+		connection->set_status(server::connection::not_found);
+		connection->set_headers(boost::make_iterator_range(headers, headers + 2));
+		connection->write("Error: Sensor id not found.\n");
+	}
+	catch(const std::out_of_range &e) {
+		connection->set_status(server::connection::no_content);
+		connection->set_headers(boost::make_iterator_range(headers, headers + 2));
+		connection->write("Error: Sensor unavailable.\n");
+	}
+	catch(const std::exception& e) {
+		connection->set_status(server::connection::internal_server_error);
+		connection->set_headers(boost::make_iterator_range(headers, headers + 2));
+		connection->write("Server error.\n");
+	}
+    
+  }
+};
+
+
 void mqttCallback(SimpleMQTTMessage *msg)
 {
   /*
@@ -144,6 +202,9 @@ void mqttCallback(SimpleMQTTMessage *msg)
   delete msg;
 }
 
+
+
+
 /*
  * Print usage information
  */
@@ -247,14 +308,30 @@ int main(int argc, char* const argv[]) {
       ms.setMessageCallback(mqttCallback);
       ms.start();
 
-      cout << "Server running...\n";
-
+      cout << "MQTT Server running..." << std::endl;
+      std::thread t1;
+      
+	  /*<< Creates the request handler. >>*/
+	  http_handler handler;
+
+	  /*<< Creates the server. >>*/		
+	  server::options options(handler);
+
+	  options.thread_pool(std::make_shared<boost::network::utils::thread_pool>());
+	  server server_(options.address("127.0.0.1").port("8080"));
+		
+	  /*<< Runs the server. >>*/
+	  t1 = std::thread([&server_] { server_.run(); });
+  
+	  cout << "HTTP Server running..." << std::endl;
       /*
        * Run (hopefully) forever...
        */
       keepRunning = 1;
       timeval start, end;
       double elapsed;
+      
+      cout << "Collect Agent running..." << std::endl;
 
       while(keepRunning) {
           gettimeofday(&start, NULL);
@@ -274,9 +351,14 @@ int main(int argc, char* const argv[]) {
       cout << "Stopping...\n";
 
       ms.stop();
+      cout << "MQTT Server stopped..." << std::endl;
+      server_.stop();
+      t1.join();
+      cout << "HTTP Server stopped..." << std::endl;
       delete mySensorDataStore;
       dcdbConn->disconnect();
       delete dcdbConn;
+      cout << "Collect Agent closed. Bye bye..." << std::endl;
   }
   catch (const exception& e) {
       cout << "Exception: " << e.what() << "\n";