App/src/Multicast.cpp

   1 #include "Tourist/App/Multicast.h"
   2 #include "Poco/Logger.h"
   3 #include "Tourist/Constants.h"
   4 #include "Tourist/Util/Util.h"
   5 #include "Poco/Instantiator.h"
   6
   7 using namespace Tourist::Util;
   8
   9 using Poco::Instantiator;
  10
  11 namespace Tourist{
  12 namespace App{
  13
  14
  15 Multicast::Multicast(Application *_app, int objCount)  :
  16 logger(Logger::get("Tourist.App.Multicast." + Util::toString(objCount) ))
  17 {
  18         this->app = _app;
  19         this->shouldBeRunning = true;
  20 }
  21
  22 Multicast::~Multicast()
  23 {
  24 }
  25
  26 void Multicast::run()
  27 {
  28         //Register for incoming multicast messages
  29         receiveMulticastMsg();
  30
  31         while(shouldBeRunning)
  32         {
  33                 processQueue();
  34                 Poco::Thread::sleep(500);
  35         }
  36
  37 }
  38
  39 void Multicast::stop()
  40 {
  41   this -> shouldBeRunning = false;
  42 }
  43
  44 void Multicast::processQueue()
  45 {
  46   //debug(logger, "Multicast::processQueue()");
  47
  48           //consume dispatched message
  49                 while (queue.size() != 0)
  50                 {
  51                         info(logger, "Processing multicast message");
  52                         //We Got the message.. let's process it
  53                         MessageNotification *msgNotify = dynamic_cast<MessageNotification*>
  54                                 (queue.dequeueNotification());
  55                         if (msgNotify == NULL)
  56                         {
  57                                 error(logger, "Notification for multicast  response is NULL!");
  58                                 continue; //let's see if there is any other message in queue.
  59                         }
  60
  61                         debug(logger, "Multicast Message Received. Parsing message ....");
  62
  63                         MsgMulticast *msgMulticast = dynamic_cast<MsgMulticast*> (msgNotify -> message);
  64                         debug(logger, "Multicast Initiated by: " + msgMulticast->getInitiatingPeer().toString());
  65                         RemoteNode thisNode;
  66                         app->getLocalNodeInfo(thisNode);
  67                         debug(logger, "This node is: " + thisNode.toString());
  68
  69
  70                         //Initiate Tree based Multicast
  71                         // Call localNode method findRemoteLinks with approprate flags
  72                         // in a loop send multicast event to all nodes in NodeSet
  73                         // end condition: if findRemoteLinks reuturns no node then no sendEvent & multicast finished.
  74
  75                         RemoteNode initNode = msgMulticast->getInitiatingPeer();
  76                         multicastRouting(&initNode, msgMulticast->getEventType(), msgMulticast->getStep());
  77
  78                 }
  79 }
  80
  81 void Multicast::multicastRouting(Node* source, int event_type, int step)
  82 {
  83         NodeSet audienceSet;
  84         int type = TYPE_PREFIX;
  85         int multicastCount = 0;
  86         bool status=false;
  87
  88         if (event_type >= 1 && event_type <= 4) {
  89                 status = app->searchRemoteLinks(*source, audienceSet, LOWER_LEVELS|EQUAL_LEVELS|FIND_PREFIXES|FIND_ALL);
  90         } else {
  91                 status = app->searchRemoteLinks(*source, audienceSet, LOWER_LEVELS|EQUAL_LEVELS|FIND_SUFFIXES|FIND_ALL);
  92                 type = TYPE_SUFFIX;
  93         }
  94
  95         debug(logger, status?"Audience Set Found":"No Audience Set Found");
  96
  97         if(status) {
  98         //TODO: Use variable instead of 64
  99         for (int i=step+1; i<=64; i++) {
 100                 Node *nextHop = app->getNextHop(audienceSet, source, i, type);
 101     if (nextHop == NULL || nextHop->hexstring() == source->hexstring()) {
 102                         continue;
 103                 }
 104
 105                 debug(logger, "Sending event to an audience set node ..... ");
 106           int ack = sendEvent(source, nextHop, event_type, i);
 107
 108                 if (ack > 0)
 109                         multicastCount++;
 110         }
 111         }
 112
 113 }
 114
 115
 116 void Multicast::levelChangeEvent(int newLevel, Node *localNode)
 117 {
 118   //Initiating prefix multicast
 119   int prefixEvent, suffixEvent;
 120   if (localNode->getLevel() > newLevel)
 121   {
 122     prefixEvent = EVENT_PREFIX_LEVEL_UP;
 123     suffixEvent = EVENT_SUFFIX_LEVEL_UP;
 124   }
 125   else
 126   {
 127     prefixEvent = EVENT_PREFIX_LEVEL_DOWN;
 128     suffixEvent = EVENT_SUFFIX_LEVEL_DOWN;
 129   }
 130
 131         Node *pTopNode;
 132         if(! (app -> getTopNode(*localNode, &pTopNode, PREFIX)))
 133                 error(logger, "Cannot find Top Node!");
 134         else debug(logger, "Top Node found");
 135   int prefixAck = sendEvent(localNode, pTopNode, prefixEvent, 0);
 136
 137         if(! (app -> getTopNode(*localNode, &pTopNode, SUFFIX)))
 138                 error(logger, "Cannot find Top Node!");
 139         else debug(logger, "Top Node found");
 140   int suffixAck = sendEvent(localNode, pTopNode, suffixEvent, 0);
 141
 142 }
 143
 144
 145 int Multicast::sendEvent(Node *source, Node *target, int event_type, int step)
 146 {
 147   if (source == NULL || target == NULL)
 148     return -1;
 149
 150   unsigned int seqN = 0;
 151   string log;
 152   const char* method = "eventOut_sendEvent";
 153
 154   /* ------------------ LOG [Send Event]----------------------- */
 155
 156   string eventType = "Un-Known";
 157   switch(event_type)
 158   {
 159     case EVENT_PREFIX_JOIN:
 160       eventType = "Prefix join";
 161       break;
 162
 163     case EVENT_PREFIX_DISCONNECT:
 164       eventType = "Prefix disconnect";
 165       break;
 166
 167     case EVENT_PREFIX_LEVEL_UP:
 168       eventType = "Prefix level up";
 169       break;
 170
 171     case EVENT_PREFIX_LEVEL_DOWN:
 172       eventType = "Prefix level down";
 173       break;
 174
 175     case EVENT_SUFFIX_JOIN:
 176       eventType = "Suffix join";
 177       break;
 178
 179     case EVENT_SUFFIX_DISCONNECT:
 180       eventType = "Suffix disconnect";
 181       break;
 182
 183     case EVENT_SUFFIX_LEVEL_UP:
 184       eventType = "Suffix level up";
 185       break;
 186
 187     case EVENT_SUFFIX_LEVEL_DOWN:
 188       eventType = "Suffix level down";
 189       break;
 190   }
 191
 192
 193         debug(logger,
 194                                 "source:source_level= " + source->hexstring() + ":" + Util::toString(source->getLevel()) +
 195                                 "\ntarget:target_level= " + target->hexstring() + ":" + Util::toString(target->getLevel()) +
 196                                 "\nevent:step= " + eventType + ":" + Util::toString(step) );
 197
 198
 199     /* ---------------------------------------------------------- */
 200
 201         //Get pointer to Topnode
 202
 203         RemoteNode *topNode;
 204
 205         RemoteNode *rNodePtr = (RemoteNode*) target;
 206         RemoteNode rNode = *rNodePtr;
 207
 208         int statusConn = app->getConnection(rNode, &topNode);
 209
 210         //Register Message Type
 211         MessageTypeData msgType;
 212         msgType.registerMessageObj(MessageTypeData::MULTICAST, new Instantiator<MsgMulticast, AbstractMessage>);
 213
 214         RemoteNode srcNode;
 215         app->getLocalNodeInfo(srcNode);
 216
 217         MsgMulticast *msgMulticast = new MsgMulticast("Multicast","Multicast", srcNode, event_type, step);
 218         string str;
 219         msgMulticast->getString(str);
 220
 221         //Instantiate MessageHeader object
 222         MessageHeader *msgHeader = new MessageHeader(MessageHeader::HEADER_VERSION, str.length(), MessageTypeData::MULTICAST);
 223
 224         //Send Multicast Request Message to Top Node
 225         topNode->sendMessage(msgHeader, msgMulticast);
 226
 227   return true;
 228 }
 229
 230
 231
 232 //------------------------------------------------------------------------------
 233 void Multicast::receiveMulticastMsg()
 234 {
 235         app->getConfig()->getMessageBus()->registerNotification("Multicast",
 236       NObserver<Multicast, MessageNotification>(*this, &Multicast::onMulticastMsg));
 237 }
 238
 239 void Multicast::onMulticastMsg(const AutoPtr<MessageNotification>& notification)
 240 {
 241         debug(logger, "Multicast::onMulticastMsg()");
 242
 243         MessageNotification *notificationObj = const_cast<MessageNotification*>(notification.get());
 244         queue.enqueueNotification(notificationObj);
 245
 246 }
 247
 248 } //namespace Tourist
 249 } //namespace App