Merge pull request #2 from adventurist/multithreading

Implement a thread pool to handle connections concurrently

CMakeLists.txt

@@ -1,10 +1,15 @@
 cmake_minimum_required(VERSION 2.8)
 project(ws_server)
-
-set(SOURCES "main.cpp" "socket_listener.cpp")
-set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -Wall -std=c++17")
+set(LISTENERLIB "socket_listener")
+set(THREADS_PREFER_PTHREAD_FLAG ON)
+find_package(Threads REQUIRED)
+set(SOURCES "src/main.cpp" "src/socket_listener.cpp" "src/task_queue.cpp")
+set(CMAKE_CXX_FLAGS_DEBUG "${CMAKE_CXX_FLAGS_DEBUG} -g -std=c++17 -pthread")
 add_executable(${PROJECT_NAME} ${SOURCES})
+add_library(${LISTENERLIB} ${SOURCES})
 target_include_directories(${PROJECT_NAME} PRIVATE
   "headers"
+  "headers/interface"
 )
+target_link_libraries(${PROJECT_NAME} Threads::Threads)
 

compile

@@ -0,0 +1,3 @@
+#!/usr/bin/env bash
+
+g++ -std=c++17 -g -I headers -pthread src/main.cpp src/socket_listener.cpp src/task_queue.cpp -o ws_server

headers/constants.h → headers/constants.hpp

@@ -1,7 +1,5 @@
 #ifndef __CONSTANTS_H__
 #define __CONSTANTS_H__
-#ifndef TRX_SOCKET_CONSTANTS
-#define TRX_SOCKET_CONSTANTS 1
 /**
  * Values used when attempting to open a socket
  */
@@ -13,5 +11,4 @@ const int SOCKET_OK = 0;
 const int WAIT_SOCKET_FAILURE = -1;
 const int WAIT_SOCKET_SUCCESS = 0;
 
-#endif
 #endif  // __CONSTANTS_H__

headers/interface/listen_interface.hpp

@@ -0,0 +1,24 @@
+#ifndef __LISTEN_INTERFACE_H__
+#define __LISTEN_INTERFACE_H__
+
+#include <memory>
+#include <string>
+
+/**
+ * ListenInterface
+ *
+ * A public interface whose implementation handles the receival of a character
+ * buffer assumed have been sent from a client socket connection, indicated by a
+ * file descriptor, communicating with the implementor.
+ *
+ * @interface
+ */
+class ListenInterface {
+ public:
+  virtual void onMessageReceived(int client_socket_fd,
+                                 std::weak_ptr<uint8_t[]> w_buffer_ptr,
+                                 ssize_t& size) = 0;
+  virtual void onConnectionClose(int client_socket_fd) = 0;
+};
+
+#endif  // __LISTEN_INTERFACE_H__

headers/interface/send_interface.hpp

@@ -0,0 +1,22 @@
+#ifndef __SEND_INTERFACE_H__
+#define __SEND_INTERFACE_H__
+
+#include <memory>
+#include <string>
+
+/**
+ * SendInterface
+ *
+ * A public interface whose implementation sends a buffer of characters to the
+ * socket connection indicated by the client_socket_fd (client socket file
+ * descriptor)
+ *
+ * @interface
+ */
+class SendInterface {
+ public:
+  virtual void sendMessage(int client_socket_fd,
+                           std::weak_ptr<uint8_t[]> w_buffer_ptr) = 0;
+};
+
+#endif  // __SEND_INTERFACE_H__

headers/listen_interface.h

@@ -1,11 +0,0 @@
-#ifndef __LISTEN_INTERFACE_H__
-#define __LISTEN_INTERFACE_H__
-
-#include <string>
-
-class ListenInterface {
- public:
-  virtual void onMessageReceived(int socket_id, std::string message) = 0;
-};
-
-#endif  // __LISTEN_INTERFACE_H__

headers/send_interface.h

@@ -1,11 +0,0 @@
-#ifndef __SEND_INTERFACE_H__
-#define __SEND_INTERFACE_H__
-
-#include <string>
-
-class SendInterface {
-  public:
-    virtual void sendMessage(int client_socket_fd, std::string message) = 0;
-};
-
-#endif  // __SEND_INTERFACE_H__

headers/socket_listener.h

@@ -1,58 +0,0 @@
-#ifndef __SOCKET_LISTENER_H__
-#define __SOCKET_LISTENER_H__
-// Project libraries
-#include "listen_interface.h"
-#include "send_interface.h"
-
-// System libraries
-#include <sys/socket.h>
-
-// C++ Libraries
-#include <string>
-
-#define MAX_BUFFER_SIZE (49152)
-
-class SocketListener : public ListenInterface, public SendInterface {
- public:
-  // constructor
-  SocketListener(std::string ipAddress, int port);
-
-  // destructor
-  ~SocketListener();
-
-  /**
-   * Send a message to a client socket described by its file descriptor
-   * @param[in] {int} client_socket_fd The client socket file descriptor
-   * @param[in] {std::string} The message to be sent
-   */
-  virtual void sendMessage(int client_socket_fd, std::string message) override;
-
-  /**
-   * Perform intialization work
-   */
-  bool init();
-
-  /**
-   * Main message loop
-   */
-  void run();
-
-  /**
-   * Perform any cleanup work
-   */
-  void cleanup();
-
-  virtual void onMessageReceived(int socket_id, std::string message) override;
-
- private:
-  // private methods
-  int createSocket();
-
-  int waitForConnection(int listening);
-
-  // private members
-  std::string m_ip_address;
-  int m_port;
-};
-
-#endif  // __SOCKET_LISTENER_H__

headers/socket_listener.hpp

@@ -0,0 +1,104 @@
+#ifndef __SOCKET_LISTENER_HPP__
+#define __SOCKET_LISTENER_HPP__
+
+// Project libraries
+#include <interface/listen_interface.hpp>
+#include <interface/send_interface.hpp>
+#include <task_queue.hpp>
+#include <types.hpp>
+#include <constants.hpp>
+// System libraries
+#include <sys/socket.h>
+
+// C++ Libraries
+#include <functional>
+#include <iostream>
+#include <memory>
+#include <string>
+#include <vector>
+
+/**
+ * SocketListener
+ *
+ * SocketListener is extensible to aid in architecting a socket server
+ */
+class SocketListener : public SendInterface, public ListenInterface {
+ public:
+  /* public classes whose instances are used by SocketListener */
+
+  /**
+   * MessageHandler
+   *
+   * Instances of this object type wrap a generic, self-contained function and
+   * behave as callable functions (functors)
+   * @class
+   */
+  class MessageHandler {
+   public:
+    MessageHandler(std::function<void(ssize_t)> cb) : m_cb(cb) {}
+
+    void operator()(ssize_t size) { m_cb(size); }
+
+   private:
+    std::function<void(ssize_t)> m_cb;
+  };
+  // constructor
+  SocketListener(int arg_num, char** args);
+
+  // destructor
+  ~SocketListener();
+
+  /**
+   * Send a message to a client socket described by its file descriptor
+   * @param[in] {int} client_socket_fd The client socket file descriptor
+   * @param[in] {std::string} The message to be sent
+   */
+  virtual void sendMessage(int client_socket_fd,
+                           std::weak_ptr<uint8_t[]> w_buffer_ptr) override;
+  /** overload variants */
+  void sendMessage(int client_socket_fd, char* message, bool short_message);
+
+  void sendMessage(int client_socket_fd, char* message, size_t size);
+
+  void sendMessage(int client_socket_fd, const char* message, size_t size);
+
+  MessageHandler createMessageHandler(std::function<void(ssize_t)> cb);
+  /**
+   * Perform intialization work
+   */
+  bool init();
+
+  /**
+   * Main message loop
+   */
+  void run();
+
+  /**
+   * Perform any cleanup work
+   */
+  void cleanup();
+
+ private:
+  // private methods
+  int createSocket();
+
+  virtual void onMessageReceived(int client_socket_fd,
+                                 std::weak_ptr<uint8_t[]> w_buffer_ptr,
+                                 ssize_t& size) override;
+
+  virtual void onConnectionClose(int client_socket_fd) override;
+
+  int waitForConnection(int listening);
+
+  void handleClientSocket(int client_socket_fd,
+                          SocketListener::MessageHandler message_handler,
+                          const std::shared_ptr<uint8_t[]>& s_buffer_ptr);
+
+  /* private members */
+  // Server arguments
+  std::string m_ip_address;
+  int m_port;
+  std::unique_ptr<TaskQueue> u_task_queue_ptr;
+};
+
+#endif  // __SOCKET_LISTENER_HPP__

headers/task_queue.hpp

@@ -0,0 +1,99 @@
+#include <atomic>
+#include <condition_variable>
+#include <functional>
+#include <mutex>
+#include <queue>
+#include <thread>
+
+/**
+ * TaskQueue
+ *
+ * A task queue employs a pool of worker threads who can execute arbitrary tasks
+ * @class
+ */
+class TaskQueue {
+ public:
+ /**
+ * @constructor
+ * Nothing fancy
+ */
+  TaskQueue();
+/**
+ * @destructor
+ * Make sure all worker threads detach or join before destroying TaskQueue
+ * instance
+ */
+  ~TaskQueue();
+
+/** PUBLIC METHODS **/
+
+/**
+ * initialize
+ *
+ * To be called after an instance of TaskQueue is created.
+ * @method
+ */
+  void initialize();
+/**
+ * pushToQueue
+ *
+ * Add a task to the queue
+ *
+ * @method
+ * @param[in] {std::function<void()>} A fully encapsulated template function
+ * with its own internal state
+ */
+  void pushToQueue(std::function<void()> fn);
+
+ private:
+
+/** PRIVATE METHODS **/
+/**
+ * workerLoop
+ *
+ * The loop is the essential lifecycle of the worker
+ * @method
+ */
+  void workerLoop();
+/**
+ * deployWorkers
+ *
+ * Procures workers and sets them into existing by executing the workerLoop
+ * function
+ * @method
+ */
+  void deployWorkers();
+/**
+ * detachThreads
+ *
+ * Allows threads to terminate.
+ * @method
+ * @cleanup
+ */
+  void detachThreads();
+
+/** PRIVATE MEMBERS **/
+
+  /**
+   * FIFO queue of templated function pointers
+   */
+  std::queue<std::function<void()>> m_task_queue;
+
+  /**
+   * vector of worker threads
+   */
+  std::vector<std::thread> m_thread_pool;
+
+  /**
+   * mutex for locking resources
+   */
+  std::mutex m_mutex_lock;
+  /**
+   * condition variable for controlling work execution
+   */
+  std::condition_variable pool_condition;
+  /**
+   * atomic boolean to ensure queue is handled in a thread-safe manner
+   */
+  std::atomic<bool> accepting_tasks;
+};

headers/types.hpp

@@ -0,0 +1,16 @@
+#ifndef __TYPES_H__
+#define __TYPES_H__
+
+#include <string>
+
+#define MAX_BUFFER_SIZE (4096)
+#define SMALL_BUFFER_SIZE (8192)
+
+template <typename MessageProcessor>
+void MessageHandler(MessageProcessor processor, int client_socket_fd,
+                    std::string message) {
+  processor(client_socket_fd, message);
+}
+
+#endif  //__TYPES_H__
+

main.cpp

@@ -1,13 +0,0 @@
-#include <iostream>
-#include <string>
-#include "headers/socket_listener.h"
-
-int main() {
-  SocketListener server("0.0.0.0", 9009);
-
-  if (server.init()) {
-    server.run();
-  }
-  return 0;
-}
-

socket_listener.cpp

@@ -1,163 +0,0 @@
-// Project headers
-#include "headers/socket_listener.h"
-
-#include "headers/constants.h"
-// System libraries
-#include <arpa/inet.h>
-#include <netdb.h>
-#include <string.h>
-#include <sys/socket.h>
-#include <sys/types.h>
-#include <unistd.h>
-// C++ Libraries
-#include <iostream>
-#include <string>
-
-/**
- * Constructor
- * Initialize with ip_address and port
- */
-SocketListener::SocketListener(std::string ip_address, int port)
-    : m_ip_address(ip_address), m_port(port) {}
-
-/**
- * Destructor
- * TODO: Determine if we should make buffer a class member
- */
-SocketListener::~SocketListener() { cleanup(); }
-
-/**
- * sendMessage
- * @method
- * Send a null-terminated array of characters, supplied as a const char pointer,
- * to a client socket described by its file descriptor
- */
-void SocketListener::sendMessage(int client_socket_fd, std::string message) {
-  send(client_socket_fd, message.c_str(), message.size() + 1, 0);
-}
-
-/**
- * init
- * TODO: Initialize buffer memory, if buffer is to be a class member
- */
-bool SocketListener::init() {
-  std::cout << "Initializing socket listener" << std::endl;
-  return true;
-}
-
-/**
- * run
- * @method
- * Main message loop
- * TODO: Implement multithreading
- */
-void SocketListener::run() {
-  // Declare, define and initialize a character buffer
-  char buf[MAX_BUFFER_SIZE] = {};
-  // Begin listening loop
-  while (true) {
-    // Call system to open a listening socket, and return its file descriptor
-    int listening_socket_fd = createSocket();
-
-    if (listening_socket_fd == SOCKET_ERROR) {
-      std::cout << "Socket error: shutting down server" << std::endl;
-      break;
-    }
-    // wait for a client connection and get its socket file descriptor
-    int client_socket_fd = waitForConnection(listening_socket_fd);
-
-    if (client_socket_fd != SOCKET_ERROR) {
-      // Destroy listening socket and deallocate its file descriptor. Only use
-      // the client socket now.
-      close(listening_socket_fd);
-      std::string buffer_string{};  // Initialize a string buffer
-      while (true) {
-        memset(buf, 0, MAX_BUFFER_SIZE);  // Zero the character buffer
-        int bytes_received = 0;
-        // Receive and write incoming data to buffer and return the number of
-        // bytes received
-        bytes_received =
-            recv(client_socket_fd, buf,
-                 MAX_BUFFER_SIZE - 2,  // Leave room for null-termination
-                 0);
-        buf[MAX_BUFFER_SIZE - 1] = 0;  // Null-terminate the character buffer
-        if (bytes_received > 0) {
-          buffer_string += buf;
-          std::cout << "Bytes received: " << bytes_received << "\nData: " << buf
-                    << std::endl;
-          // Handle incoming message
-          onMessageReceived(client_socket_fd, std::string(buf));
-        } else {
-          std::cout << "client disconnected" << std::endl;
-          break;
-        }
-      }
-      // Zero the buffer again before closing
-      memset(buf, 0, MAX_BUFFER_SIZE);
-      // TODO: Determine if we should free memory, or handle as class member
-      close(client_socket_fd);  // Destroy client socket and deallocate its fd
-    }
-  }
-}
-
-/**
- * cleanUp
- * @method
- * TODO: Determine if we should be cleaning up buffer memory
- */
-void SocketListener::cleanup() { std::cout << "Cleaning up" << std::endl; }
-/**
- * createSocket
- * Open a listening socket and return its file descriptor
- */
-int SocketListener::createSocket() {
-  /* Call the system to open a socket passing arguments for
-   ipv4 family, tcp type and no additional protocol info */
-  int listening_socket_fd = socket(AF_INET, SOCK_STREAM, 0);
-
-  if (listening_socket_fd != SOCKET_ERROR) {
-    // Create socket structure to hold address and type
-    sockaddr_in socket_struct;
-    socket_struct.sin_family = AF_INET;  // ipv4
-    socket_struct.sin_port =
-        htons(m_port);  // convert byte order of port value from host to network
-    inet_pton(AF_INET, m_ip_address.c_str(),  // convert address to binary
-              &socket_struct.sin_addr);
-    // Bind local socket address to socket file descriptor
-    int bind_result = bind(
-        listening_socket_fd,         // TODO: Use C++ cast on next line?
-        (sockaddr *)&socket_struct,  // cast socket_struct to more generic type
-        sizeof(socket_struct));
-    if (bind_result != SOCKET_ERROR) {
-      // Listen for connections to socket and allow up to max number of
-      // connections for queue
-      int listen_result = listen(listening_socket_fd, SOMAXCONN);
-      if (listen_result == SOCKET_ERROR) {
-        return WAIT_SOCKET_FAILURE;
-      }
-    } else {
-      return WAIT_SOCKET_FAILURE;
-    }
-  }
-  return listening_socket_fd;  // Return socket file descriptor
-}
-/**
- * waitForConnection
- * @method
- * Takes first connection on queue of pending connections, creates a new socket
- * and returns its file descriptor
- */
-int SocketListener::waitForConnection(int listening_socket) {
-  int client_socket_fd = accept(listening_socket, NULL, NULL);
-  return client_socket_fd;
-}
-/**
- * onMessageReceived
- * @method
- * @override
- * Handle messages successfully received from a client socket
- */
-void SocketListener::onMessageReceived(int socket_id, std::string message) {
-  sendMessage(socket_id, message);
-}
-

socket_server

@@ -0,0 +1,379 @@
+# Doxyfile 1.8.16
+
+#---------------------------------------------------------------------------
+# Project related configuration options
+#---------------------------------------------------------------------------
+DOXYFILE_ENCODING      = UTF-8
+PROJECT_NAME           = "Socket Server"
+PROJECT_NUMBER         =
+PROJECT_BRIEF          =
+PROJECT_LOGO           =
+OUTPUT_DIRECTORY       = docs
+CREATE_SUBDIRS         = NO
+ALLOW_UNICODE_NAMES    = NO
+OUTPUT_LANGUAGE        = English
+OUTPUT_TEXT_DIRECTION  = None
+BRIEF_MEMBER_DESC      = YES
+REPEAT_BRIEF           = YES
+ABBREVIATE_BRIEF       = "The $name class" \
+                         "The $name widget" \
+                         "The $name file" \
+                         is \
+                         provides \
+                         specifies \
+                         contains \
+                         represents \
+                         a \
+                         an \
+                         the
+ALWAYS_DETAILED_SEC    = NO
+INLINE_INHERITED_MEMB  = NO
+FULL_PATH_NAMES        = YES
+STRIP_FROM_PATH        =
+STRIP_FROM_INC_PATH    =
+SHORT_NAMES            = NO
+JAVADOC_AUTOBRIEF      = NO
+JAVADOC_BANNER         = NO
+QT_AUTOBRIEF           = NO
+MULTILINE_CPP_IS_BRIEF = NO
+INHERIT_DOCS           = YES
+SEPARATE_MEMBER_PAGES  = NO
+TAB_SIZE               = 4
+ALIASES                =
+TCL_SUBST              =
+OPTIMIZE_OUTPUT_FOR_C  = NO
+OPTIMIZE_OUTPUT_JAVA   = YES
+OPTIMIZE_FOR_FORTRAN   = NO
+OPTIMIZE_OUTPUT_VHDL   = NO
+OPTIMIZE_OUTPUT_SLICE  = NO
+EXTENSION_MAPPING      =
+MARKDOWN_SUPPORT       = YES
+TOC_INCLUDE_HEADINGS   = 5
+AUTOLINK_SUPPORT       = YES
+BUILTIN_STL_SUPPORT    = NO
+CPP_CLI_SUPPORT        = NO
+SIP_SUPPORT            = NO
+IDL_PROPERTY_SUPPORT   = YES
+DISTRIBUTE_GROUP_DOC   = NO
+GROUP_NESTED_COMPOUNDS = NO
+SUBGROUPING            = YES
+INLINE_GROUPED_CLASSES = NO
+INLINE_SIMPLE_STRUCTS  = NO
+TYPEDEF_HIDES_STRUCT   = NO
+LOOKUP_CACHE_SIZE      = 0
+#---------------------------------------------------------------------------
+# Build related configuration options
+#---------------------------------------------------------------------------
+EXTRACT_ALL            = YES
+EXTRACT_PRIVATE        = NO
+EXTRACT_PRIV_VIRTUAL   = NO
+EXTRACT_PACKAGE        = NO
+EXTRACT_STATIC         = NO
+EXTRACT_LOCAL_CLASSES  = YES
+EXTRACT_LOCAL_METHODS  = NO
+EXTRACT_ANON_NSPACES   = NO
+HIDE_UNDOC_MEMBERS     = NO
+HIDE_UNDOC_CLASSES     = NO
+HIDE_FRIEND_COMPOUNDS  = NO
+HIDE_IN_BODY_DOCS      = NO
+INTERNAL_DOCS          = NO
+CASE_SENSE_NAMES       = YES
+HIDE_SCOPE_NAMES       = NO
+HIDE_COMPOUND_REFERENCE= NO
+SHOW_INCLUDE_FILES     = YES
+SHOW_GROUPED_MEMB_INC  = NO
+FORCE_LOCAL_INCLUDES   = NO
+INLINE_INFO            = YES
+SORT_MEMBER_DOCS       = YES
+SORT_BRIEF_DOCS        = NO
+SORT_MEMBERS_CTORS_1ST = NO
+SORT_GROUP_NAMES       = NO
+SORT_BY_SCOPE_NAME     = NO
+STRICT_PROTO_MATCHING  = NO
+GENERATE_TODOLIST      = YES
+GENERATE_TESTLIST      = YES
+GENERATE_BUGLIST       = YES
+GENERATE_DEPRECATEDLIST= YES
+ENABLED_SECTIONS       =
+MAX_INITIALIZER_LINES  = 30
+SHOW_USED_FILES        = YES
+SHOW_FILES             = YES
+SHOW_NAMESPACES        = YES
+FILE_VERSION_FILTER    =
+LAYOUT_FILE            =
+CITE_BIB_FILES         =
+#---------------------------------------------------------------------------
+# Configuration options related to warning and progress messages
+#---------------------------------------------------------------------------
+QUIET                  = NO
+WARNINGS               = YES
+WARN_IF_UNDOCUMENTED   = YES
+WARN_IF_DOC_ERROR      = YES
+WARN_NO_PARAMDOC       = NO
+WARN_AS_ERROR          = NO
+WARN_FORMAT            = "$file:$line: $text"
+WARN_LOGFILE           =
+#---------------------------------------------------------------------------
+# Configuration options related to the input files
+#---------------------------------------------------------------------------
+INPUT                  =
+INPUT_ENCODING         = UTF-8
+FILE_PATTERNS          = *.c \
+                         *.cc \
+                         *.cxx \
+                         *.cpp \
+                         *.c++ \
+                         *.java \
+                         *.ii \
+                         *.ixx \
+                         *.ipp \
+                         *.i++ \
+                         *.inl \
+                         *.idl \
+                         *.ddl \
+                         *.odl \
+                         *.h \
+                         *.hh \
+                         *.hxx \
+                         *.hpp \
+                         *.h++ \
+                         *.cs \
+                         *.d \
+                         *.php \
+                         *.php4 \
+                         *.php5 \
+                         *.phtml \
+                         *.inc \
+                         *.m \
+                         *.markdown \
+                         *.md \
+                         *.mm \
+                         *.dox \
+                         *.py \
+                         *.pyw \
+                         *.f90 \
+                         *.f95 \
+                         *.f03 \
+                         *.f08 \
+                         *.f \
+                         *.for \
+                         *.tcl \
+                         *.vhd \
+                         *.vhdl \
+                         *.ucf \
+                         *.qsf \
+                         *.ice
+RECURSIVE              = YES
+EXCLUDE                =
+EXCLUDE_SYMLINKS       = NO
+EXCLUDE_PATTERNS       =
+EXCLUDE_SYMBOLS        =
+EXAMPLE_PATH           =
+EXAMPLE_PATTERNS       = *
+EXAMPLE_RECURSIVE      = NO
+IMAGE_PATH             =
+INPUT_FILTER           =
+FILTER_PATTERNS        =
+FILTER_SOURCE_FILES    = NO
+FILTER_SOURCE_PATTERNS =
+USE_MDFILE_AS_MAINPAGE =
+#---------------------------------------------------------------------------
+# Configuration options related to source browsing
+#---------------------------------------------------------------------------
+SOURCE_BROWSER         = NO
+INLINE_SOURCES         = NO
+STRIP_CODE_COMMENTS    = YES
+REFERENCED_BY_RELATION = NO
+REFERENCES_RELATION    = NO
+REFERENCES_LINK_SOURCE = YES
+SOURCE_TOOLTIPS        = YES
+USE_HTAGS              = NO
+VERBATIM_HEADERS       = YES
+#---------------------------------------------------------------------------
+# Configuration options related to the alphabetical class index
+#---------------------------------------------------------------------------
+ALPHABETICAL_INDEX     = YES
+COLS_IN_ALPHA_INDEX    = 5
+IGNORE_PREFIX          =
+#---------------------------------------------------------------------------
+# Configuration options related to the HTML output
+#---------------------------------------------------------------------------
+GENERATE_HTML          = YES
+HTML_OUTPUT            = html
+HTML_FILE_EXTENSION    = .html
+HTML_HEADER            =
+HTML_FOOTER            =
+HTML_STYLESHEET        =
+HTML_EXTRA_STYLESHEET  =
+HTML_EXTRA_FILES       =
+HTML_COLORSTYLE_HUE    = 220
+HTML_COLORSTYLE_SAT    = 100
+HTML_COLORSTYLE_GAMMA  = 80
+HTML_TIMESTAMP         = NO
+HTML_DYNAMIC_MENUS     = YES
+HTML_DYNAMIC_SECTIONS  = NO
+HTML_INDEX_NUM_ENTRIES = 100
+GENERATE_DOCSET        = NO
+DOCSET_FEEDNAME        = "Doxygen generated docs"
+DOCSET_BUNDLE_ID       = org.doxygen.Project
+DOCSET_PUBLISHER_ID    = org.doxygen.Publisher
+DOCSET_PUBLISHER_NAME  = Publisher
+GENERATE_HTMLHELP      = NO
+CHM_FILE               =
+HHC_LOCATION           =
+GENERATE_CHI           = NO
+CHM_INDEX_ENCODING     =
+BINARY_TOC             = NO
+TOC_EXPAND             = NO
+GENERATE_QHP           = NO
+QCH_FILE               =
+QHP_NAMESPACE          = org.doxygen.Project
+QHP_VIRTUAL_FOLDER     = doc
+QHP_CUST_FILTER_NAME   =
+QHP_CUST_FILTER_ATTRS  =
+QHP_SECT_FILTER_ATTRS  =
+QHG_LOCATION           =
+GENERATE_ECLIPSEHELP   = NO
+ECLIPSE_DOC_ID         = org.doxygen.Project
+DISABLE_INDEX          = NO
+GENERATE_TREEVIEW      = NO
+ENUM_VALUES_PER_LINE   = 4
+TREEVIEW_WIDTH         = 250
+EXT_LINKS_IN_WINDOW    = NO
+FORMULA_FONTSIZE       = 10
+FORMULA_TRANSPARENT    = YES
+USE_MATHJAX            = NO
+MATHJAX_FORMAT         = HTML-CSS
+MATHJAX_RELPATH        = https://cdnjs.cloudflare.com/ajax/libs/mathjax/2.7.5/
+MATHJAX_EXTENSIONS     =
+MATHJAX_CODEFILE       =
+SEARCHENGINE           = YES
+SERVER_BASED_SEARCH    = NO
+EXTERNAL_SEARCH        = NO
+SEARCHENGINE_URL       =
+SEARCHDATA_FILE        = searchdata.xml
+EXTERNAL_SEARCH_ID     =
+EXTRA_SEARCH_MAPPINGS  =
+#---------------------------------------------------------------------------
+# Configuration options related to the LaTeX output
+#---------------------------------------------------------------------------
+GENERATE_LATEX         = NO
+LATEX_OUTPUT           =
+LATEX_CMD_NAME         =
+MAKEINDEX_CMD_NAME     = makeindex
+LATEX_MAKEINDEX_CMD    = makeindex
+COMPACT_LATEX          = NO
+PAPER_TYPE             = a4
+EXTRA_PACKAGES         =
+LATEX_HEADER           =
+LATEX_FOOTER           =
+LATEX_EXTRA_STYLESHEET =
+LATEX_EXTRA_FILES      =
+PDF_HYPERLINKS         = YES
+USE_PDFLATEX           = YES
+LATEX_BATCHMODE        = NO
+LATEX_HIDE_INDICES     = NO
+LATEX_SOURCE_CODE      = NO
+LATEX_BIB_STYLE        = plain
+LATEX_TIMESTAMP        = NO
+LATEX_EMOJI_DIRECTORY  =
+#---------------------------------------------------------------------------
+# Configuration options related to the RTF output
+#---------------------------------------------------------------------------
+GENERATE_RTF           = NO
+RTF_OUTPUT             = rtf
+COMPACT_RTF            = NO
+RTF_HYPERLINKS         = NO
+RTF_STYLESHEET_FILE    =
+RTF_EXTENSIONS_FILE    =
+RTF_SOURCE_CODE        = NO
+#---------------------------------------------------------------------------
+# Configuration options related to the man page output
+#---------------------------------------------------------------------------
+GENERATE_MAN           = NO
+MAN_OUTPUT             = man
+MAN_EXTENSION          = .3
+MAN_SUBDIR             =
+MAN_LINKS              = NO
+#---------------------------------------------------------------------------
+# Configuration options related to the XML output
+#---------------------------------------------------------------------------
+GENERATE_XML           = NO
+XML_OUTPUT             = xml
+XML_PROGRAMLISTING     = YES
+XML_NS_MEMB_FILE_SCOPE = NO
+#---------------------------------------------------------------------------
+# Configuration options related to the DOCBOOK output
+#---------------------------------------------------------------------------
+GENERATE_DOCBOOK       = NO
+DOCBOOK_OUTPUT         = docbook
+DOCBOOK_PROGRAMLISTING = NO
+#---------------------------------------------------------------------------
+# Configuration options for the AutoGen Definitions output
+#---------------------------------------------------------------------------
+GENERATE_AUTOGEN_DEF   = NO
+#---------------------------------------------------------------------------
+# Configuration options related to the Perl module output
+#---------------------------------------------------------------------------
+GENERATE_PERLMOD       = NO
+PERLMOD_LATEX          = NO
+PERLMOD_PRETTY         = YES
+PERLMOD_MAKEVAR_PREFIX =
+#---------------------------------------------------------------------------
+# Configuration options related to the preprocessor
+#---------------------------------------------------------------------------
+ENABLE_PREPROCESSING   = YES
+MACRO_EXPANSION        = NO
+EXPAND_ONLY_PREDEF     = NO
+SEARCH_INCLUDES        = YES
+INCLUDE_PATH           =
+INCLUDE_FILE_PATTERNS  =
+PREDEFINED             =
+EXPAND_AS_DEFINED      =
+SKIP_FUNCTION_MACROS   = YES
+#---------------------------------------------------------------------------
+# Configuration options related to external references
+#---------------------------------------------------------------------------
+TAGFILES               =
+GENERATE_TAGFILE       =
+ALLEXTERNALS           = NO
+EXTERNAL_GROUPS        = YES
+EXTERNAL_PAGES         = YES
+#---------------------------------------------------------------------------
+# Configuration options related to the dot tool
+#---------------------------------------------------------------------------
+CLASS_DIAGRAMS         = YES
+DIA_PATH               =
+HIDE_UNDOC_RELATIONS   = YES
+HAVE_DOT               = NO
+DOT_NUM_THREADS        = 0
+DOT_FONTNAME           = Helvetica
+DOT_FONTSIZE           = 10
+DOT_FONTPATH           =
+CLASS_GRAPH            = YES
+COLLABORATION_GRAPH    = YES
+GROUP_GRAPHS           = YES
+UML_LOOK               = NO
+UML_LIMIT_NUM_FIELDS   = 10
+TEMPLATE_RELATIONS     = NO
+INCLUDE_GRAPH          = YES
+INCLUDED_BY_GRAPH      = YES
+CALL_GRAPH             = NO
+CALLER_GRAPH           = NO
+GRAPHICAL_HIERARCHY    = YES
+DIRECTORY_GRAPH        = YES
+DOT_IMAGE_FORMAT       = png
+INTERACTIVE_SVG        = NO
+DOT_PATH               =
+DOTFILE_DIRS           =
+MSCFILE_DIRS           =
+DIAFILE_DIRS           =
+PLANTUML_JAR_PATH      =
+PLANTUML_CFG_FILE      =
+PLANTUML_INCLUDE_PATH  =
+DOT_GRAPH_MAX_NODES    = 50
+MAX_DOT_GRAPH_DEPTH    = 0
+DOT_TRANSPARENT        = NO
+DOT_MULTI_TARGETS      = NO
+GENERATE_LEGEND        = YES
+DOT_CLEANUP            = YES

src/main.cpp

@@ -0,0 +1,21 @@
+#include <iostream>
+#include <string>
+
+#include "../headers/socket_listener.hpp"
+
+/** \mainpage
+ * SocketListener constructor takes 2 parameters (std::string ip, int port).
+ *
+ * Calling the "run()" method will cause it to for and handle multiple
+ * concurrent socket connections.
+ */
+
+int main(int argc, char** argv) {
+  SocketListener server(argc, argv);
+  if (server.init()) {
+    std::cout << "Running message loop" << std::endl;
+    server.run();
+  }
+  return 0;
+}
+

src/socket_listener.cpp

@@ -0,0 +1,245 @@
+// Project headers
+#include <socket_listener.hpp>
+// System libraries
+#include <arpa/inet.h>
+#include <netdb.h>
+#include <string.h>
+#include <sys/socket.h>
+#include <sys/types.h>
+#include <unistd.h>
+// C++ Libraries
+#include <iostream>
+#include <memory>
+#include <string>
+
+/**
+ * Constructor
+ * Initialize with ip_address, port and message_handler
+ */
+SocketListener::SocketListener(int arg_num, char** args) : m_port(-1) {
+  for (int i = 0; i < arg_num; i++) {
+    std::string argument = std::string(args[i]);
+    std::cout << args[i] << std::endl;
+    if (argument.find("--ip") != -1) {
+      m_ip_address = argument.substr(5);
+      continue;
+    }
+    if (argument.find("--port") != -1) {
+      m_port = std::stoi(argument.substr(7));
+      continue;
+    }
+    if (m_ip_address.empty()) {
+      m_ip_address = "0.0.0.0";
+    }
+    if (m_port == -1) {
+      m_port = 9009;
+    }
+  }
+}
+
+/**
+ * Destructor
+ * TODO: Determine if we should make buffer a class member
+ */
+SocketListener::~SocketListener() { cleanup(); }
+
+SocketListener::MessageHandler SocketListener::createMessageHandler(
+    std::function<void(ssize_t)> cb) {
+  return MessageHandler(cb);
+}
+
+void SocketListener::onMessageReceived(int client_socket_fd,
+                                       std::weak_ptr<uint8_t[]> w_buffer_ptr,
+                                       ssize_t& size) {
+  std::cout << "This should be overridden" << std::endl;
+  sendMessage(client_socket_fd, w_buffer_ptr);
+}
+
+void SocketListener::onConnectionClose(int client_socket_fd) {
+  std::cout << "This should be overridden" << std::endl;
+}
+
+/**
+ * sendMessage
+ * @method
+ * Send a null-terminated array of characters, supplied as a const uint8_t
+ * pointer, to a client socket described by its file descriptor
+ */
+void SocketListener::sendMessage(int client_socket_fd,
+                                 std::weak_ptr<uint8_t[]> w_buffer_ptr) {
+  std::shared_ptr<uint8_t[]> s_buffer_ptr = w_buffer_ptr.lock();
+  if (s_buffer_ptr) {
+    send(client_socket_fd, s_buffer_ptr.get(),
+         static_cast<size_t>(MAX_BUFFER_SIZE) + 1, 0);
+  } else {
+    std::cout << "Could not send message to client " << client_socket_fd
+              << ". Buffer does not exist." << std::endl;
+  }
+}
+
+void SocketListener::sendMessage(int client_socket_fd, char* message,
+                                 bool short_message) {
+  if (short_message) {
+    send(client_socket_fd, message, static_cast<size_t>(SMALL_BUFFER_SIZE) + 1,
+         0);
+  } else {
+    send(client_socket_fd, message, static_cast<size_t>(MAX_BUFFER_SIZE) + 1,
+         0);
+  }
+}
+
+void SocketListener::sendMessage(int client_socket_fd, char buffer[],
+                                 size_t size) {
+  send(client_socket_fd, buffer, size + 1, 0);
+}
+
+void SocketListener::sendMessage(int client_socket_fd, const char* buffer,
+                                 size_t size) {
+  send(client_socket_fd, buffer, size + 1, 0);
+}
+
+/**
+ * init
+ * TODO: Initialize buffer memory, if buffer is to be a class member
+ */
+bool SocketListener::init() {
+  std::cout << "Initializing socket listener" << std::endl;
+  u_task_queue_ptr = std::make_unique<TaskQueue>();
+  u_task_queue_ptr->initialize();
+  return true;
+}
+
+void SocketListener::handleClientSocket(
+    int client_socket_fd, SocketListener::MessageHandler message_handler,
+    const std::shared_ptr<uint8_t[]>& s_buffer_ptr) {
+  for (;;) {
+    memset(s_buffer_ptr.get(), 0,
+           MAX_BUFFER_SIZE);  // Zero the character buffer
+    // Receive and write incoming data to buffer and return the number of
+    // bytes received
+    ssize_t size = recv(client_socket_fd, s_buffer_ptr.get(),
+                        MAX_BUFFER_SIZE,  // Leave room for null-termination ?
+                        0);
+    //    s_buffer_ptr.get()[MAX_BUFFER_SIZE - 1] =
+    //        0;  // Null-terminate the character buffer
+    if (size > 0) {
+      std::cout << "Client " << client_socket_fd << "\nBytes received: " << size
+                << "\nData: " << std::hex << s_buffer_ptr.get() << std::endl;
+      // Handle incoming message
+      message_handler(size);
+    } else {
+      std::cout << "Client " << client_socket_fd << " disconnected"
+                << std::endl;
+      onConnectionClose(client_socket_fd);
+      // Zero the buffer again before closing
+      memset(s_buffer_ptr.get(), 0, MAX_BUFFER_SIZE);
+      break;
+    }
+  }
+  // TODO: Determine if we should free memory, or handle as class member
+  close(client_socket_fd);  // Destroy client socket and deallocate its fd
+}
+
+/**
+ * run
+ * @method
+ * Main message loop
+ * TODO: Implement multithreading
+ */
+void SocketListener::run() {
+  // Begin listening loop
+  while (true) {
+    std::cout << "Begin" << std::endl;
+    // Call system to open a listening socket, and return its file descriptor
+    int listening_socket_fd = createSocket();
+
+    if (listening_socket_fd == SOCKET_ERROR) {
+      std::cout << "Socket error: shutting down server" << std::endl;
+      break;
+    }
+    std::cout << "Attempting to wait for connection" << std::endl;
+    // wait for a client connection and get its socket file descriptor
+    int client_socket_fd = waitForConnection(listening_socket_fd);
+
+    if (client_socket_fd != SOCKET_ERROR) {
+      // Destroy listening socket and deallocate its file descriptor. Only use
+      // the client socket now.
+      close(listening_socket_fd);
+      {
+        std::shared_ptr<uint8_t[]> s_buffer_ptr(new uint8_t[MAX_BUFFER_SIZE]);
+        std::weak_ptr<uint8_t[]> w_buffer_ptr(s_buffer_ptr);
+        // TODO: Stop the use of this size variable, by changing the
+        // specification of handleClientSocket
+        std::function<void(ssize_t)> message_send_fn =
+            [this, client_socket_fd, w_buffer_ptr](ssize_t size) {
+              this->onMessageReceived(client_socket_fd, w_buffer_ptr, size);
+            };
+        MessageHandler message_handler = createMessageHandler(message_send_fn);
+        std::cout << "Pushing client to queue" << std::endl;
+        u_task_queue_ptr->pushToQueue(
+            std::bind(&SocketListener::handleClientSocket, this,
+                      client_socket_fd, message_handler,
+                      std::forward<std::shared_ptr<uint8_t[]>>(s_buffer_ptr)));
+      }
+    }
+  }
+}
+
+/**
+ * cleanUp
+ * @method
+ * TODO: Determine if we should be cleaning up buffer memory
+ */
+void SocketListener::cleanup() { std::cout << "Cleaning up" << std::endl; }
+/**
+ * createSocket
+ * Open a listening socket and return its file descriptor
+ */
+int SocketListener::createSocket() {
+  /* Call the system to open a socket passing arguments for
+   ipv4 family, tcp type and no additional protocol info */
+  int listening_socket_fd = socket(AF_INET, SOCK_STREAM, 0);
+
+  if (listening_socket_fd != SOCKET_ERROR) {
+    std::cout << "Created listening socket" << std::endl;
+    // Create socket structure to hold address and type
+    sockaddr_in socket_struct;
+    socket_struct.sin_family = AF_INET;  // ipv4
+    socket_struct.sin_port =
+        htons(m_port);  // convert byte order of port value from host to network
+    inet_pton(AF_INET, m_ip_address.c_str(),  // convert address to binary
+              &socket_struct.sin_addr);
+
+    int socket_option = 1;
+    // Free up the port to begin listening again
+    setsockopt(listening_socket_fd, SOL_SOCKET, SO_REUSEADDR, &socket_option,
+               sizeof(socket_option));
+
+    // Bind local socket address to socket file descriptor
+    int bind_result = bind(
+        listening_socket_fd,        // TODO: Use C++ cast on next line?
+        (sockaddr*)&socket_struct,  // cast socket_struct to more generic type
+        sizeof(socket_struct));
+    if (bind_result != SOCKET_ERROR) {
+      // Listen for connections to socket and allow up to max number of
+      // connections for queue
+      int listen_result = listen(listening_socket_fd, SOMAXCONN);
+      if (listen_result == SOCKET_ERROR) {
+        return WAIT_SOCKET_FAILURE;
+      }
+    } else {
+      return WAIT_SOCKET_FAILURE;
+    }
+  }
+  return listening_socket_fd;  // Return socket file descriptor
+}
+/**
+ * waitForConnection
+ * @method
+ * Takes first connection on queue of pending connections, creates a new
+ * socket and returns its file descriptor
+ */
+int SocketListener::waitForConnection(int listening_socket) {
+  int client_socket_fd = accept(listening_socket, NULL, NULL);
+  return client_socket_fd;
+}

src/task_queue.cpp

@@ -0,0 +1,112 @@
+#include <task_queue.hpp>
+
+#include <iostream>
+
+/**
+ * @global
+ * {int} num_threads A rough estimate as to the number of threads we can run
+ * concurrently
+ */
+int num_threads = std::thread::hardware_concurrency();
+
+/**
+ * @constructor
+ * Nothing fancy
+ */
+TaskQueue::TaskQueue() {}
+/**
+ * @destructor
+ * Make sure all worker threads detach or join before destroying TaskQueue
+ * instance
+ */
+TaskQueue::~TaskQueue() { detachThreads(); }
+
+/**
+ * pushToQueue
+ *
+ * Add a task to the queue
+ *
+ * @method
+ * @param[in] {std::function<void()>} A fully encapsulated template function
+ * with its own internal state
+ */
+void TaskQueue::pushToQueue(std::function<void()> fn) {
+  std::unique_lock<std::mutex> lock(m_mutex_lock);  // obtain mutex
+  m_task_queue.push(fn);                            // add work to queue
+  lock.unlock();
+  pool_condition.notify_one();  // one worker can begin waiting to perform work
+}
+
+/**
+ * workerLoop
+ *
+ * The loop is the essential lifecycle of the worker
+ * @method
+ */
+void TaskQueue::workerLoop() {
+  std::function<void()> fn;
+  for (;;) {
+    {  // encapsulate atomic management of queue
+      std::unique_lock<std::mutex> lock(m_mutex_lock);  // obtain mutex
+      pool_condition
+          .wait(  // condition: not accepting tasks or queue is not empty
+              lock,
+              [this]() { return !accepting_tasks || !m_task_queue.empty(); });
+      std::cout << "Wait condition met" << std::endl;
+      if (!accepting_tasks && m_task_queue.empty()) {
+        // If the queue is empty, it's safe to begin accepting tasks
+        accepting_tasks = true;
+        continue;
+      }
+      std::cout << "Taking task" << std::endl;
+      fn = m_task_queue.front();  // obtain task from FIFO container
+      m_task_queue.pop();
+      accepting_tasks = true;  // begin accepting before lock expires
+    }                          // queue management complete (lock expires)
+    fn();                      // work
+  }
+}
+
+/**
+ * deployWorkers
+ *
+ * Procures workers and sets them into existing by executing the workerLoop
+ * function
+ * @method
+ */
+void TaskQueue::deployWorkers() {
+  for (int i = 0; i < (num_threads - 1); i++) {
+    m_thread_pool.push_back(std::thread([this]() { workerLoop(); }));
+  }
+  // TODO: mutex may not be necessary, as accepting_tasks is atomic
+  std::unique_lock<std::mutex> lock(m_mutex_lock);  // obtain mutex
+  accepting_tasks = false;  // allow pool wait condition to be met
+  lock.unlock();
+  // when we send the notification immediately, the consumer will try to get the
+  // lock , so unlock asap
+  pool_condition.notify_all();
+}  // lock expires
+
+/**
+ * initialize
+ *
+ * To be called after an instance of TaskQueue is created.
+ * @method
+ */
+void TaskQueue::initialize() { deployWorkers(); }
+
+/**
+ * detachThreads
+ *
+ * Allows threads to terminate.
+ * @method
+ * @cleanup
+ */
+void TaskQueue::detachThreads() {
+  for (std::thread& t : m_thread_pool) {
+    if (t.joinable()) {
+      t.detach();
+    }
+  }
+}
+