Implements rudimentary thread-safety for EventBus and Logger classes.

src/core/EventBus.cpp

@@ -5,21 +5,28 @@ namespace Core {
 
 // ============================== EventBus Class ==============================
 
-EventBus::EventBus(Logger& logger) : _logger(logger) {
+EventBus::EventBus(Logger& logger) :
+    _logger(logger)
+{
     _logger.debug.ln ("Starting Event Bus...");
     _thread = thread(&EventBus::run, this);
+    _logger.debug.ln ("Event Bus Started");
 }
 
 EventBus::~EventBus() {
+    _logger.debug.ln ("Stopping Event Bus...");
     _thread.join();
     // destroy all events in _events vector
-    _logger.debug.ln ("EventBus has been shut down.");
+    for (auto event : _events) { delete event; }
+    _logger.debug.ln ("Event Bus stopped");
 }
 
 Event& EventBus::registerEvent(const string& eventID) {
+    _mutex.lock();
     for ( const auto event : _events ) {
         if (event->getID() == eventID) {
             _logger.debug << "Event '" << eventID << "' found" << Logger::commit;
+            _mutex.unlock();
             return *event;
         }
     }
@@ -27,27 +34,28 @@ Event& EventBus::registerEvent(const string& eventID) {
     _logger.debug << "Registering new Event '" << eventID << "'" << Logger::commit;
     Event* event = new Event(eventID);
     _events.push_back(event);
+    _mutex.unlock();
     return *event;
 }
 
 void EventBus::run() {
-    _logger.warn.ln("Hi from EventBus run method! From within Event Thread! Dont forget to make Logger class thread-safe!");
+    _logger.warn.ln("Hi from EventBus run method, from within Event Thread!");
 }
 
 // =============================== Event Class ================================
 
 Event::Event(const string id) : _id(id) {}
 
-Event::~Event() {
+Event::~Event() {}
-
-}
 
 const string& Event::getID() const {
     return _id;
 };
 
 void Event::subscribe(EventSubscriber* es) {
+    _mutex.lock();
     _subscribers.push_back(es);
+    _mutex.unlock();
 }
 
 void Event::fire() {

src/core/Logger.cpp

@@ -63,17 +63,19 @@ Logger::Sink::Sink (const Level level, Logger& parent) :
     _lineIsPrefixed(false) {}
 
 void Logger::Sink::ln  ( const string& msg ) {
-    ln(msg.c_str());
+    ln (msg.c_str());
 }
 
 void Logger::Sink::ln  ( const char* msg ) {
+    _parent._mutex.lock();
     if ( _level >= _parent._logLevel ) {
         // If line has already been started, abort it and start new one:
         if(_lineIsPrefixed) *_parent._output << endl;
         prefixLine(true);
-        *_parent._output <<  msg << endl;
+        *(_parent._output) <<  msg << endl;
         cout << msg << CC_RESET "\n";
     }
+    _parent._mutex.unlock();
 }
 
 void Logger::Sink::prefixLine(bool resetToFalse) {
@@ -82,19 +84,19 @@ void Logger::Sink::prefixLine(bool resetToFalse) {
         switch ( _level)
         {
             case DEBUG:
-                *_parent._output << timestamp << " " << "DEBUG  ";
+                *(_parent._output) << timestamp << " " << "DEBUG  ";
                 cout << timestamp << " " << CC_BLUE "DEBUG  ";
                 break;
             case INFO:
-                *_parent._output << timestamp << " " << "INFO   ";
+                *(_parent._output) << timestamp << " " << "INFO   ";
                 cout << timestamp << " " << CC_GREEN "INFO   ";
                 break;
             case WARN:
-                *_parent._output << timestamp << " " << "WARN   ";
+                *(_parent._output) << timestamp << " " << "WARN   ";
                 cout << timestamp << " " << CC_YELLOW "WARN   ";
                 break;
             case FATAL:
-                *_parent._output << timestamp << " " << "FATAL  ";
+                *(_parent._output) << timestamp << " " << "FATAL  ";
                 cout << timestamp << " " << CC_RED "FATAL  ";
                 break;
         }
@@ -103,61 +105,68 @@ void Logger::Sink::prefixLine(bool resetToFalse) {
 }
 
 Logger::Sink& Logger::Sink::operator<<(const string& msg) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         prefixLine();
-        *_parent._output << msg;
+        *(_parent._output) << msg;
         cout << msg;
     }
     return *this;
 }
 
 Logger::Sink& Logger::Sink::operator<<(const char* msg) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         prefixLine();
-        *_parent._output << msg;
+        *(_parent._output) << msg;
         cout << msg;
     }
     return *this;
 }
 
 Logger::Sink& Logger::Sink::operator<<(const int msg) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         prefixLine();
-        *_parent._output << msg;
+        *(_parent._output) << msg;
         cout << msg;
     }
     return *this;
 }
 
 Logger::Sink& Logger::Sink::operator<<(const long msg) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         prefixLine();
-        *_parent._output << msg;
+        *(_parent._output) << msg;
         cout << msg;
     }
     return *this;
 }
 
 Logger::Sink& Logger::Sink::operator<<(const unsigned long msg) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         prefixLine();
-        *_parent._output << msg;
+        *(_parent._output) << msg;
         cout << msg;
     }
     return *this;
 }
 
 Logger::Sink& Logger::Sink::operator<<(const char character) {
+    _parent._mutex.try_lock();
     if ( _level >= _parent._logLevel ) {
         switch(character) {
             // COMMIT - End message line
             case 0:
-              *_parent._output << endl;
+              *(_parent._output) << endl;
               cout << CC_RESET "\n";
               _lineIsPrefixed = false;
             break;
         }
     }
+    _parent._mutex.unlock();
     return *this;
 }
 

src/core/include/EventBus.hpp

@@ -5,6 +5,7 @@
 #include <string>
 #include <vector>
 #include <memory>
+#include <mutex>
 
 namespace Plains {
 namespace Core {
@@ -12,6 +13,7 @@ namespace Core {
 using std::thread;
 using std::string;
 using std::vector;
+using std::mutex;
 using std::unique_ptr;
 
 class EventSubscriber {
@@ -30,6 +32,7 @@ class Event {
     private:
     const string _id;
     vector<EventSubscriber*> _subscribers;
+    mutex _mutex;
 };
 
 class EventBus {
@@ -38,12 +41,14 @@ class EventBus {
     ~EventBus();
     Event& registerEvent(const string& eventID);
     
+    private:
+    void run();
+
     private:
     Logger& _logger;
     thread _thread;
     vector<Event*> _events;
-
+    mutex _mutex;
-    void run();
 };
 
 }}

src/core/include/Logger.hpp

@@ -2,6 +2,7 @@
 #define LOGGER_HPP
 #include <string>
 #include <fstream>
+#include <mutex>
 
 #define CC(COLOR) CC_##COLOR
 #define CC_RESET "\033[0m"
@@ -20,6 +21,7 @@ namespace Core {
 
 using std::string;
 using std::ofstream;
+using std::mutex;
 //using std::ostream;
 //using std::streambuf;
 
@@ -39,6 +41,7 @@ class Logger {
         Logger& _parent;
         const Logger::Level _level;
         void prefixLine(bool resetToFalse = false);
+        
         public:
         Sink(const Level level, Logger& parent);
         void ln (const string& msg);
@@ -56,6 +59,7 @@ class Logger {
     private:
     enum Level _logLevel;
     ofstream* _output;
+    mutex _mutex;
     static string genTimeStamp();
 };
 

src/core/main.cpp

@@ -3,6 +3,7 @@
 #include "DNS.hpp"
 #include <vector>
 #include <string>
+#include <exception>
 
 using namespace Plains;
 using namespace Core;
@@ -15,13 +16,13 @@ int main(int args, char **argv) {
 
     logger.info.ln ("Starting PlaiNS...");
 
-    try
-    {
     vector<string> arguments;
     for ( unsigned index = 0; index < args; ++index ) {
         arguments.push_back (argv[index]);
     }
 
+    try
+    {
         ConfigManager    cfg(logger, arguments);
         EventBus         eventBus(logger);
         Resolver::Server resolver(logger, eventBus);
@@ -56,13 +57,15 @@ int main(int args, char **argv) {
         // Run something on main thread...
         logger.info.ln ("Shutting down PlaiNS. Good Bye!");  
     }
-    catch (Exception& e)
+    catch (Exception& e) {
-    {
         logger.fatal << "Uncaught Exception: " << e.what() << Logger::commit;
         return 1;
     }
-    catch (int e)
+    catch (std::system_error& e) {
-    {
+        logger.fatal << "Uncaught system_error exception: " << e.what() << Logger::commit;
+        return 1;
+    }
+    catch (int e) {
         logger.fatal << "Uncaught Exception with error code " << e << Logger::commit;
         return 1;
     }

src/resolver/Server.cpp

@@ -12,6 +12,7 @@ namespace Plains {
 namespace Resolver {
 
 using asio::ip::udp;
+using asio::ip::tcp;
 
 static char* byteToHex(char c) {
     static char* out = new char[3];
@@ -93,9 +94,12 @@ Server::Server(Logger& logger, EventBus& eventBus)
     _udpSocket = new udp::socket(_ioContext, udp::endpoint(udp::v4(), 8853));
     _ioContext.run();
     doReceive();
+    _logger.debug.ln ("Resolver started");
 }
 
-Server::~Server() {}
+Server::~Server() {
+    _udpSocket->close();
+}
 
 //void Server::run() {
 //    _ioContext.run();
@@ -129,4 +133,6 @@ void Server::doSend(std::size_t length)
 
 
 
+
+
 }}