MessageHub.cpp

#include "MessageHub.h"
 
namespace Cogs::Network {
 
  MessageHub::HubList& allHubs() {
    static MessageHub::HubList hubs;
    return hubs;
  }
 
  Mutex& allHubsMutex() {
    static Mutex mutex;
    return mutex;
  }
 
  MessageHub* find(uint32_t hubid) {
    LockGuard             g(allHubsMutex());
    MessageHub::HubList&  hubs = allHubs();
 
    for (auto i = hubs.begin(), e = hubs.end(); i != e; ++i ) {
      if ((*i)->getID() == hubid) {
        return *i;
      }
    }
    return nullptr;
  }
}
 
Cogs::Network::MessageHub::MessageHub(uint32_t ident) {
  id = ident;
 
  LockGuard   g(allHubsMutex());
 
  allHubs().push_back(this);
}
 
Cogs::Network::MessageHub::~MessageHub() {
  LockGuard g(allHubsMutex());
  HubList&  hubs = allHubs();
 
  for (auto i = hubs.begin(), e = hubs.end(); i != e; ++i ) {
    if (*i == this) {
      hubs.erase(i);
      break;
    }
  }
  disconnectFromAllSenders();
  removeAllListeners();
  flushQueuedMessages();
}
 
bool Cogs::Network::MessageHub::addListener(MessageHub* hub, bool bidirectional) {
  if (hub) {
          bool  exists = false;
 
          hubsMutex.lock();
    for (auto i = listeners.begin(), e = listeners.end(); i != e; ++i) {
      if (*i == hub) {
        exists = true;
        break;
      }
    }
    if (!exists)  {
      listeners.push_back(hub);
    }
    hubsMutex.unlock();
 
    if (bidirectional) {
      hub->addListener(this, false);
    }
 
    if (!exists) {
      hub->addSender(this);
      return true;
    }
  }
  return false;
}
 
bool Cogs::Network::MessageHub::addListener(uint32_t hubid, bool bidirectional) {
  return addListener(find(hubid), bidirectional);
}
 
bool Cogs::Network::MessageHub::listenTo(MessageHub* hub, bool bidirectional) {
  return hub ? hub->addListener(this, bidirectional) : false;
}
 
bool Cogs::Network::MessageHub::listenTo(uint32_t hubid, bool bidirectional) {
  return listenTo(find(hubid), bidirectional);
}
 
bool Cogs::Network::MessageHub::removeListener(MessageHub* hub) {
  if (hub) {
          bool  removed = false;
 
          hubsMutex.lock();
    for (auto i = listeners.begin(), e = listeners.end(); i != e; ++i) {
      if (*i == hub) {
        listeners.erase(i);
        removed = true;
        break;
      }
    }
          hubsMutex.unlock();
 
          if (removed) {
                  hub->removeSender(this);
                  return true;
          }
  }
        return false;
}
 
bool Cogs::Network::MessageHub::removeListener(uint32_t hubid) {
  return removeListener(find(hubid));
}
 
void Cogs::Network::MessageHub::removeAllListeners() {
  HubList  copy;
 
  hubsMutex.lock();
  std::swap(listeners, copy);
  hubsMutex.unlock();
 
  for (auto i = copy.begin(), e = copy.end(); i != e; ++i) {
    (*i)->removeSender(this);
  }
}
 
void Cogs::Network::MessageHub::disconnectFromSender(uint32_t hubid) {
  HubList copy;
 
  hubsMutex.lock();
  copy = senders;
  hubsMutex.unlock();
 
  for (auto i = copy.begin(), e = copy.end(); i != e; ++i) {
    if ((*i)->getID() == hubid) {
      (*i)->removeListener(this);
      break;
    }
  }
}
 
void Cogs::Network::MessageHub::disconnectFromAllSenders() {
  HubList copy;
 
  hubsMutex.lock();
  std::swap(senders, copy);
  hubsMutex.unlock();
 
  for (auto i = copy.begin(), e = copy.end(); i != e; ++i) {
    (*i)->removeListener(this);
  }
}
 
void Cogs::Network::MessageHub::broadcastMessage(const Message::Ptr& message) {
  LockGuard g(allHubsMutex());
  HubList&  hubs = allHubs();
 
  for (auto i = hubs.begin(), e = hubs.end(); i != e; ++i ) {
    if (*i != this) {
      (*i)->queueMessage(message);
    }
  }
}
 
void Cogs::Network::MessageHub::sendMessage(const Message::Ptr& message) {
  LockGuard g(hubsMutex);
 
  for (auto i = listeners.begin(), e = listeners.end(); i != e; ++i) {
    (*i)->queueMessage(message);
  }
}
 
bool Cogs::Network::MessageHub::processMessages(size_t limit) {
  size_t  queuedMessages = getNoOfQueuedMessages();
  bool    ret = queuedMessages ? true : false;
 
  if (queuedMessages > (limit * 10)) {
    limit = noLimit;
  }
 
  for (Message::Ptr message = getNextMessage(); message; message = getNextMessage()) {
    Message reader(message->getID(), *message);
 
    processMessage(reader);
 
    if (!--limit) {
      break;
    }
  }
  return ret;
}
 
void Cogs::Network::MessageHub::queueMessage(const Message::Ptr& message) {
  if (!isSendOnly()) {
    LockGuard g(messagesMutex);
 
    messages.push_back(message);
  }
}
 
size_t Cogs::Network::MessageHub::getNoOfQueuedMessages() {
  LockGuard g(messagesMutex);
 
  return messages.size();
}
 
void Cogs::Network::MessageHub::flushQueuedMessages() {
  LockGuard g(messagesMutex);
 
  messages.clear();
}
 
Cogs::Network::Message::Ptr Cogs::Network::MessageHub::getNextMessage() {
  LockGuard g(messagesMutex);
 
  if (!messages.empty()) {
    Message::Ptr  message = messages.front();
 
    messages.pop_front();
    return message;
  }
  return nullptr;
}
 
void Cogs::Network::MessageHub::addSender(MessageHub* sender) {
        LockGuard g(hubsMutex);
 
  for (auto i = senders.begin(), e = senders.end(); i != e; ++i) {
    if (*i == sender) {
      return;
    }
  }
  senders.push_back(sender);
}
 
void Cogs::Network::MessageHub::removeSender(const MessageHub* sender) {
        LockGuard g(hubsMutex);
 
  for (auto i = senders.begin(), e = senders.end(); i != e; ++i) {
    if (*i == sender) {
      senders.erase(i);
      break;
    }
  }
}