fixes issues with nulls in protobufs resulting in lopped messages (#10)

* add more logging to event dispatcher

* add more logging to message dispatcher

* Adds more log messages

* fixes issues with nulls in protobufs resulting in lopped messages

Author: thoraxe

src/Events/EventDispatcher.cpp

@@ -23,6 +23,7 @@
 #include "../Commands/JoinSecurityCommand.h"
 #include "../Commands/LeaveSecurityCommand.h"
 #include "../Shared/FactoryT.h"
+#include "../Logging/loguru.hpp"
 #include <Poco/Delegate.h>
 #include <assert.h>
 
@@ -107,13 +108,15 @@ EventDispatcher::~EventDispatcher()
 // Helper(s)
 void EventDispatcher::Enqueue(Message* pMessage)
 {
+    LOG_SCOPE_FUNCTION(8);
     m_anEventQueueMutex.lock();
     m_anEventQueue.push(pMessage);
     m_anEventQueueMutex.unlock();
 }
 
 Message* EventDispatcher::Dequeue()
 {
+    LOG_SCOPE_FUNCTION(8);
     Message* pMessage = NULL;
 
     m_anEventQueueMutex.lock();
@@ -126,6 +129,8 @@ Message* EventDispatcher::Dequeue()
 
 GameEventBuffer* EventDispatcher::CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType eEntityGameEvent_EntityGameEventBufferType, AEntity* pEntity)
 {
+    LOG_SCOPE_FUNCTION(8);
+    LOG_F(8, "Entity Game Event");
     EntityGameEvent_Dependencies anEntityGameEvent_Dependencies(eEntityGameEvent_EntityGameEventBufferType, pEntity);
     GameEventBuffer* pGameEvent = m_anEntityGameEventFactory.Create(anEntityGameEvent_Dependencies);
 
@@ -134,6 +139,8 @@ GameEventBuffer* EventDispatcher::CreateGameEvent(EntityGameEventBuffer_EntityGa
 
 GameEventBuffer* EventDispatcher::CreateGameEvent(SecurityGameEventBuffer_SecurityGameEventBufferType eSecurityGameEvent_SecurityGameEventBufferType, const std::string& strUUID)
 {
+    LOG_SCOPE_FUNCTION(8);
+    LOG_F(8, "Security Game Event");
     SecurityGameEvent_Dependencies aSecurityGameEvent_Dependencies(eSecurityGameEvent_SecurityGameEventBufferType, strUUID);
     GameEventBuffer* pGameEvent = m_aSecurityGameEventFactory.Create(aSecurityGameEvent_Dependencies);
 
@@ -143,6 +150,7 @@ GameEventBuffer* EventDispatcher::CreateGameEvent(SecurityGameEventBuffer_Securi
 // Dispatches all the events it has received to it's listeners
 void EventDispatcher::Dispatch()
 {
+    LOG_SCOPE_FUNCTION(8);
     Message* pMessage = NULL;
     m_anEventQueueMutex.lock();
     while (!m_anEventQueue.empty())
@@ -157,75 +165,87 @@ void EventDispatcher::Dispatch()
 // Entity event response
 void EventDispatcher::HandlePodCreatedEvent(const void* pSender, Pod*& pPod)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_CREATE, static_cast<AEntity*>(pPod));
     Enqueue(pGameEvent);
 }
 
 void EventDispatcher::HandlePodUpdatedEvent(const void* pSender, Pod*& pPod)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_UPDATE, static_cast<AEntity*>(pPod));
     Enqueue(pGameEvent);}
 
 void EventDispatcher::HandlePodDestroyedEvent(const void* pSender, Pod*& pPod)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_DESTROY, static_cast<AEntity*>(pPod));
     Enqueue(pGameEvent);}
 
 void EventDispatcher::HandleBulletCreatedEvent(const void* pSender, Bullet*& pBullet)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_CREATE, static_cast<AEntity*>(pBullet));
     Enqueue(pGameEvent);
 }
 
 void EventDispatcher::HandleBulletUpdatedEvent(const void* pSender, Bullet*& pBullet)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_UPDATE, static_cast<AEntity*>(pBullet));
     Enqueue(pGameEvent);
 }
 
 void EventDispatcher::HandleBulletDestroyedEvent(const void* pSender, Bullet*& pBullet)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(EntityGameEventBuffer_EntityGameEventBufferType_DESTROY, static_cast<AEntity*>(pBullet));
     Enqueue(pGameEvent);
 }
 
 // Event Consumer event response
 void EventDispatcher::HandleJoinSecurityCommandFactoryCreatedEvent(const void* pSender, JoinSecurityCommand*& pJoinSecurityCommand)
 {
+    LOG_SCOPE_FUNCTION(8);
     assert(pJoinSecurityCommand);
 
     pJoinSecurityCommand->ExecutedEvent += Poco::Delegate<EventDispatcher, const std::string&>(this, &EventDispatcher::HandleJoinSecurityCommandExecutedEvent);
 }
 
 void EventDispatcher::HandleJoinSecurityCommandFactoryDestroyedEvent(const void* pSender, JoinSecurityCommand*& pJoinSecurityCommand)
 {
+    LOG_SCOPE_FUNCTION(8);
     assert(pJoinSecurityCommand);
 
     pJoinSecurityCommand->ExecutedEvent -= Poco::Delegate<EventDispatcher, const std::string&>(this, &EventDispatcher::HandleJoinSecurityCommandExecutedEvent);
 }
 
 void EventDispatcher::HandleLeaveSecurityCommandFactoryCreatedEvent(const void* pSender, LeaveSecurityCommand*& pLeaveSecurityCommand)
 {
+    LOG_SCOPE_FUNCTION(8);
     assert(pLeaveSecurityCommand);
 
     pLeaveSecurityCommand->ExecutedEvent += Poco::Delegate<EventDispatcher, const std::string&>(this, &EventDispatcher::HandleLeaveSecurityCommandExecutedEvent);
 }
 
 void EventDispatcher::HandleLeaveSecurityCommandFactoryDestroyedEvent(const void* pSender, LeaveSecurityCommand*& pLeaveSecurityCommand)
 {
+    LOG_SCOPE_FUNCTION(8);
     assert(pLeaveSecurityCommand);
 
     pLeaveSecurityCommand->ExecutedEvent -= Poco::Delegate<EventDispatcher, const std::string&>(this, &EventDispatcher::HandleLeaveSecurityCommandExecutedEvent);
 }
 
 void EventDispatcher::HandleJoinSecurityCommandExecutedEvent(const void* pSender, const std::string& strUUID)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(SecurityGameEventBuffer_SecurityGameEventBufferType_JOIN, strUUID);
     Enqueue(pGameEvent);
 }
 
 void EventDispatcher::HandleLeaveSecurityCommandExecutedEvent(const void* pSender, const std::string& strUUID)
 {
+    LOG_SCOPE_FUNCTION(8);
     GameEventBuffer* pGameEvent = CreateGameEvent(SecurityGameEventBuffer_SecurityGameEventBufferType_LEAVE, strUUID);
     Enqueue(pGameEvent);
 }

src/Network/MessageDispatcher.cpp

@@ -74,6 +74,7 @@ void MessageDispatcher::Enqueue(std::pair<const unsigned char*, unsigned long>*
 
 void MessageDispatcher::Enqueue(google::protobuf::Message* pEventMessage)
 {
+    LOG_SCOPE_FUNCTION(4);
     assert(pEventMessage);
 
     std::pair<const unsigned char*, unsigned long>* pMessagePair = MessageToPair(pEventMessage);
@@ -131,6 +132,7 @@ std::pair<const unsigned char*, unsigned long>* MessageDispatcher::MessageToPair
 // via the configured simple async producer
 void MessageDispatcher::Dispatch()
 {
+    int x = 1;
     LOG_SCOPE_FUNCTION(3);
     m_aMessageQueueMutex.lock();
     while (!m_aMessageQueue.empty())
@@ -143,18 +145,23 @@ void MessageDispatcher::Dispatch()
             {
                 // TODO: Proton TESTME
 //                proton::message msg((const unsigned char*)pMessagePair->first);
-                proton::binary* body = new proton::binary(std::string((const char*)pMessagePair->first));
+                std::string pre_proton = (const char*)(pMessagePair->first);
+                LOG_F(8, "pMessagePair length(): %i", pre_proton.length());
+                LOG_F(8, "pMessagePair second(): %i", pMessagePair->second);
+                proton::binary* body = new proton::binary(std::string((const char*)pMessagePair->first, pMessagePair->second));
                 proton::message msg(*body);
 //                msg->body(*body);
                 msg.content_type("proton::BINARY");
                 msg.content_encoding("proton::BINARY");
 
+                LOG_F(7, "message #: %i", x);
                 LOG_F(7, "proton msg body type is %s", proton::type_name(msg.body().type()).c_str());
                 LOG_F(7, "proton msg content_type is %s", msg.content_type().c_str());
                 LOG_F(8, "proton msg content_encoding is %s", msg.content_encoding().c_str());
                 LOG_F(7, "proton msg body is %s", proton::coerce<std::string>(msg.body()).c_str());
 
                 m_psender->send(msg);
+                x++;
             }
         }
         catch ( std::exception& e )