L:/channel/channel/include/TrieRouter.h

Go to the documentation of this file.

// Copyright (c) 2005, 2006 Yigong Liu
// Permission to use, copy, modify, distribute and sell this software for any 
//     purpose is hereby granted without fee, provided that the above copyright 
//     notice appear in all copies and that both that copyright notice and this 
//     permission notice appear in supporting documentation.
// The author makes no representations about the 
//     suitability of this software for any purpose. It is provided "as is" 
//     without express or implied warranty.

#ifndef _TRIE_ROUTER_H_
#define _TRIE_ROUTER_H_

//ACE headers
#include "ace/Log_Msg.h"
//for ACE_GUARD macros, need the following 2 headers
#include "ace/Synch.h"
#include "ace/OS.h"

//std
#include <map>
#include <list>
#include <vector>
#include <string>
#include <algorithm>

//Channel
#include <BaseDef.h>
#include <RouterBase.h>
#include <Msg.h>

namespace channel {

  template <class, class, class, class> class Channel;

  template <class IdType, class IdTrait, class SynchPolicy>
  class TrieRouter : public RouterBase {
  public:
    enum { NamespaceType = NAMESPACE_HIERARCHICAL };

    typedef Channel<IdType, IdTrait, SynchPolicy, TrieRouter<IdType, IdTrait, SynchPolicy> > Channel;
    typedef Destination<Channel, NamespaceType> Destination;
    typedef Source<Channel, NamespaceType> Source;
    //cannot use Channel::Msg here, since Channel is not defined yet
    //Channel is defined using Router
    typedef Msg<IdType, SynchPolicy> Msg;
    typedef PubSub_Info_Msg<IdType> PubSub_Info_Msg;
    typedef typename IdTrait::NameType NameType;

  private:

    struct Pub_Registry {
      std::list<Source *> *pubers_;
      std::list<Source *> *w_pubers_;
      typename SynchPolicy::RW_MUTEX pubers_lock_;
      Pub_Registry() {
        pubers_ = NULL;
        w_pubers_ = NULL;
      }
      ~Pub_Registry() {
        ACE_WRITE_GUARD(typename SynchPolicy::RW_MUTEX, guard, pubers_lock_);
        if (pubers_ != NULL) {
          delete pubers_;
        }
        if (w_pubers_ != NULL) {
          delete w_pubers_;
        }
      }
    };
    struct Sub_Registry {
      std::map<Destination *, PubSub_Scope> *subers_;
      std::map<Destination *, PubSub_Scope> *w_subers_;
      typename SynchPolicy::RW_MUTEX subers_lock_;
      Sub_Registry() {
        subers_ = NULL;
        w_subers_ = NULL;
      }
      ~Sub_Registry() {
        ACE_WRITE_GUARD(typename SynchPolicy::RW_MUTEX, guard, subers_lock_);
        if (subers_ != NULL) {
          delete subers_;
        }
        if (w_subers_ != NULL) {
          delete w_subers_;
        }
      }
    };

    template <class Registry>
    struct TrieNode {
      NameType name;
      TrieNode *parent;
      std::map<NameType, TrieNode *> children;
      typename SynchPolicy::RW_MUTEX children_lock_;
      Registry registry;
      TrieNode(NameType n, TrieNode *p) : name(n), parent(p) {
      }
      ~TrieNode() {
        ACE_WRITE_GUARD(typename SynchPolicy::RW_MUTEX, guard, children_lock_);
        for(typename std::map<NameType, TrieNode *>::iterator iter = children.begin();
            iter != children.end(); iter++) {
          if (iter->second != NULL)
            delete iter->second;
        }
      }
    };

    //namespace - pub/sub info indexed by IdType
    TrieNode<Sub_Registry> *sub_trie_root_;
    TrieNode<Pub_Registry> *pub_trie_root_;

  public:

    TrieRouter() {
      sub_trie_root_ = new TrieNode<Sub_Registry>(IdTrait::RootName, NULL);
      pub_trie_root_ = new TrieNode<Pub_Registry>(IdTrait::RootName, NULL);
    }

    ~TrieRouter() {
      delete sub_trie_root_;
      delete pub_trie_root_;
    }

    void propagate_child_pub_registry (IdType t, TrieNode<Pub_Registry> *node, Oper_Type oper) {
      Pub_Registry *reg = &(node->registry);
      //if encounter wildcard, stop going deeper
      {
        ACE_READ_GUARD(typename SynchPolicy::RW_MUTEX, guard, reg->pubers_lock_);
        if (reg->w_pubers_ != NULL && !reg->w_pubers_->empty()) {
          t.push_back(IdTrait::WildcardName);
          propagate_change_to_neighbors(oper, t);
          t.pop_back();
          return;
        }
        if (reg->pubers_ != NULL && !reg->pubers_->empty()) {
          propagate_change_to_neighbors(oper, t);
        }
      }
      //recursively update children
      ACE_READ_GUARD(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_);
      for(typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.begin();
          iter != node->children.end(); iter++) {
        t.push_back(iter->first);
        propagate_child_pub_registry (t, iter->second, oper);
        t.pop_back();
      }
    };


    void propagate_child_sub_registry (IdType t, TrieNode<Sub_Registry> *node, Oper_Type oper) {
      bool send2remote = false;
      Sub_Registry *reg = &(node->registry);
      //if encounter wildcard, stop going deeper
      {
        ACE_READ_GUARD(typename SynchPolicy::RW_MUTEX, guard, reg->subers_lock_);
        if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
          for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
              iter != reg->w_subers_->end() && !send2remote; iter++)
            if (iter->first->type() == MEMBER_LOCAL && 
                (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
              send2remote = true;
          if (send2remote) {
            t.push_back(IdTrait::WildcardName);
            propagate_change_to_neighbors(oper, t);
            t.pop_back();
            return;
          }
        }
        if (reg->subers_ != NULL && !reg->subers_->empty()) {
          for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->begin();
              iter != reg->subers_->end() && !send2remote; iter++)
            if (iter->first->type() == MEMBER_LOCAL &&
                (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
              send2remote = true;
          if(send2remote)
            propagate_change_to_neighbors(oper, t);
        }
      }
      //recursively update children
      ACE_READ_GUARD(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_);
      for(typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.begin();
          iter != node->children.end(); iter++) {
        t.push_back(iter->first);
        propagate_child_sub_registry (t, iter->second, oper);
        t.pop_back();
      }
    };



    Status publish_msg(IdType t, PubSub_Scope c, Source *src)
      {
        bool send2remote = false;
        Pub_Registry *reg = NULL;
        TrieNode<Pub_Registry> *node = pub_trie_root_;
        bool wildcard = false;
        bool hidden = false;
        int sz = t.size();

        if (!(c == SCOPE_REMOTE || c == SCOPE_GLOBAL)) 
          return SUCCESS;

        for (int i=0; i<sz; i++) {
          if (t[i] == IdTrait::WildcardName) {
            if (i == (sz-1)) {
              wildcard = true;
              break;
            } else {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: Wildcard in middle [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
          }
          //check if it is hidden
          if (!hidden) {
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_pubers_ != NULL && !reg->w_pubers_->empty()) 
              hidden = true;
          }
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
          typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.find(t[i]);
          if (iter == node->children.end()) {
            TrieNode<Pub_Registry> *cnd = new TrieNode<Pub_Registry> (t[i], node);
            node->children[t[i]] = cnd;
            node = cnd;
          } 
          else
            node = iter->second;
        }

        //update pub registry
        if (!wildcard) {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->pubers_ == NULL) {
              reg->pubers_ = new std::list<Source *>();
              send2remote = true;
            }
            reg->pubers_->push_back(src);
          }
          //broadcast t to remote
          if (send2remote && !hidden && src->type() == MEMBER_LOCAL)
            propagate_change_to_neighbors(OPER_PUBLISH, t);
        }
        else {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_pubers_ == NULL) {
              reg->w_pubers_ = new std::list<Source *>();
              send2remote = true;
            }
            reg->w_pubers_->push_back(src);
          }
          if (send2remote && !hidden && src->type() == MEMBER_LOCAL) {
            //unpublish child registry
            IdType id = t;
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
            for(typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.begin();
                iter != node->children.end(); iter++) {
              id[sz-1] = iter->first;
              propagate_child_pub_registry (id, iter->second, OPER_UNPUBLISH);
            }
            //broadcast t to remote
            propagate_change_to_neighbors(OPER_PUBLISH, t);
          }
        }
        return SUCCESS;
      }

    Status unpublish_msg(IdType t, PubSub_Scope s, Source *src)
      {
        bool send2remote = false;
        Pub_Registry *reg = NULL;
        TrieNode<Pub_Registry> *node = pub_trie_root_;
        bool wildcard = false;
        bool hidden = false;
        int sz = t.size();

        if (!(s == SCOPE_REMOTE || s == SCOPE_GLOBAL)) 
          return SUCCESS;

        for (int i=0; i<sz; i++) {
          if (t[i] == IdTrait::WildcardName) {
            if (i == (sz-1)) {
              wildcard = true;
              break;
            } else {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: Wildcard in middle [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
          }
          //check if it is hidden
          if (!hidden) {
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_pubers_ != NULL && !reg->w_pubers_->empty()) 
              hidden = true;
          }
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
          typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.find(t[i]);
          if (iter == node->children.end()) {
            ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: not found [%s]\n",ID2STR(t).c_str()));
            return FAILURE;
          }
          node = iter->second;
        }
        if (!wildcard) {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->pubers_ == NULL) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            typename std::list<Source *>::iterator iter = std::find(reg->pubers_->begin(),
                 reg->pubers_->end(), src);
            if (iter == reg->pubers_->end()) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            reg->pubers_->erase(iter);
            if (reg->pubers_->empty()) {
              send2remote = true;
            }
          }
          //broadcast t to remote
          if (send2remote && !hidden && src->type() == MEMBER_LOCAL)
            propagate_change_to_neighbors(OPER_UNPUBLISH, t);
        }
        else {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.pubers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_pubers_ == NULL) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            typename std::list<Source *>::iterator iter = std::find(reg->w_pubers_->begin(),
                 reg->w_pubers_->end(), src);
            if (iter == reg->w_pubers_->end()) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unpublish_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            reg->w_pubers_->erase(iter);
            if (reg->w_pubers_->empty()) {
              send2remote = true;
            }
          }
          //
          if (send2remote && !hidden && src->type() == MEMBER_LOCAL) {
            //broadcast t to remote
            propagate_change_to_neighbors(OPER_UNPUBLISH, t);
            //publish child registry
            IdType id = t;
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
            for(typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.begin();
                iter != node->children.end(); iter++) {
              id[sz-1] = iter->first;
              propagate_child_pub_registry (id, iter->second, OPER_PUBLISH);
            }
          }
        }
        return SUCCESS;
      }

    Status subscribe_msg(IdType t, PubSub_Scope c, Destination *dest)
      {
        bool send2remote = false;
        Sub_Registry *reg = NULL;
        TrieNode<Sub_Registry> *node = sub_trie_root_;
        bool wildcard = false;
        bool hidden = false;
        int sz = t.size();

        for (int i=0; i<sz; i++) {
          if (t[i] == IdTrait::WildcardName) {
            if (i == (sz-1)) {
              wildcard = true;
              break;
            } else {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::subscribe_msg: Wildcard in middle [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
          }
          if (!hidden) {
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                  iter != reg->w_subers_->end() && !hidden; iter++) 
                if (iter->first->type() == MEMBER_LOCAL && 
                    (iter->second == SCOPE_GLOBAL || iter->second == SCOPE_REMOTE))
                  hidden = true;
            }
          }
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
          typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.find(t[i]);
          if (iter == node->children.end()) {
            TrieNode<Sub_Registry> *cnd = new TrieNode<Sub_Registry> (t[i], node);
            node->children[t[i]] = cnd;
            node = cnd;
          } 
          else
            node = iter->second;
        }
        //update subscriptions
        //if !wildcard, add src & st -> node.registry->subers, return
        if (!wildcard) {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->subers_ == NULL) {
              reg->subers_ = new std::map<Destination *, PubSub_Scope>();
              send2remote = true;
            }
            if (!send2remote && !hidden) {
              send2remote = true;
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->begin();
                  iter != reg->subers_->end(); iter++)
                if (iter->first->type() == MEMBER_LOCAL &&
                    (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
                  send2remote = false;
            }
            (*reg->subers_)[dest] = c;
          }
          //broadcast t to remote
          if (send2remote && !hidden && dest->type() == MEMBER_LOCAL && 
              (c == SCOPE_REMOTE || c == SCOPE_GLOBAL)) {
            propagate_change_to_neighbors(OPER_SUBSCRIBE, t);
          }
        }
        else {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_subers_ == NULL) {
              reg->w_subers_ = new std::map<Destination *, PubSub_Scope>();
              send2remote = true;
            }
            if (!send2remote && !hidden) {
              send2remote = true;
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                  iter != reg->w_subers_->end(); iter++)
                if (iter->first->type() == MEMBER_LOCAL && 
                    (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
                  send2remote = false;
            }
            (*reg->w_subers_)[dest] = c;
          }
          //broadcast t to remote
          if (send2remote && !hidden && dest->type() == MEMBER_LOCAL && 
              (c == SCOPE_REMOTE || c == SCOPE_GLOBAL)) {
            //unsubscribe child registry
            IdType id = t;
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
            for(typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.begin();
                iter != node->children.end(); iter++) {
              id[sz-1] = iter->first;
              propagate_child_sub_registry (id, iter->second, OPER_UNSUBSCRIBE);
            }
            propagate_change_to_neighbors(OPER_SUBSCRIBE, t);
          }
        }
        return SUCCESS;
      }

    Status unsubscribe_msg(IdType t, Destination *dest)
      {
        bool send2remote = false;
        Sub_Registry *reg = NULL;
        TrieNode<Sub_Registry> *node = sub_trie_root_;
        bool wildcard = false;
        bool hidden = false;
        int sz = t.size();

        for (int i=0; i<sz; i++) {
          if (t[i] == IdTrait::WildcardName) {
            if (i == (sz-1)) {
              wildcard = true;
              break;
            } else {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: Wildcard in middle [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
          }
          if (!hidden) {
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                  iter != reg->w_subers_->end() && !hidden; iter++) 
                if (iter->first->type() == MEMBER_LOCAL && 
                    (iter->second == SCOPE_GLOBAL || iter->second == SCOPE_REMOTE))
                  hidden = true;
            }
          }
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
          typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.find(t[i]);
          if (iter == node->children.end()) {
            ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: not found [%s]\n",ID2STR(t).c_str()));
            return FAILURE;
          }
          node = iter->second;
        }
        //update subscriptions
        if (!wildcard) {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->subers_ == NULL) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->find(dest);
            if (iter == reg->subers_->end()) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            if (iter->first->type() == MEMBER_LOCAL && 
                (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
              send2remote = true;
            reg->subers_->erase(dest);
            if (send2remote && !hidden && !reg->subers_->empty()) {
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->begin();
                  iter != reg->subers_->end(); iter++)
                if (iter->first->type() == MEMBER_LOCAL && 
                    (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
                  send2remote = false;
            }
          }
          //broadcast t to remote
          if (send2remote && !hidden)
            propagate_change_to_neighbors(OPER_UNSUBSCRIBE, t);
        }
        else {
          {
            ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_subers_ == NULL) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->find(dest);
            if (iter == reg->w_subers_->end()) {
              ACE_DEBUG((LM_DEBUG, "TrieRouter::unsubscribe_msg: not found [%s]\n",ID2STR(t).c_str()));
              return FAILURE;
            }
            if (iter->first->type() == MEMBER_LOCAL && 
                (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
              send2remote = true;
            reg->w_subers_->erase(dest);
            if (send2remote && !hidden && !reg->w_subers_->empty()) {
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                  iter != reg->w_subers_->end(); iter++)
                if (iter->first->type() == MEMBER_LOCAL && 
                    (iter->second == SCOPE_REMOTE || iter->second == SCOPE_GLOBAL))
                  send2remote = false;
            }
          }
          //broadcast t to remote
          if (send2remote && !hidden) {
            propagate_change_to_neighbors(OPER_UNSUBSCRIBE, t);
            //subscribe child registry
            IdType id = t;
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
            for(typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.begin();
                iter != node->children.end(); iter++) {
              id[sz-1] = iter->first;
              propagate_child_sub_registry (id, iter->second, OPER_SUBSCRIBE);
            }
          }
        }
        return SUCCESS;
      }

    Status route_msg(Msg *msg, Member_Type src_type, PubSub_Scope pub_scope, ACE_Time_Value *timeout=0)
      {
        Sub_Registry *reg = NULL;
        TrieNode<Sub_Registry> *node = sub_trie_root_;
        Msg *dup=NULL;
        int sz = msg->type.size();

        for (int i=0; i<sz; i++) {
          {
            //check if any wildcard subscription
            ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);
            reg = &(node->registry);
            if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
              for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                  iter != reg->w_subers_->end(); iter++) {
                //dispatch to wildcard subscriber
                short src_row = src_type * SCOPE_NUMBER + pub_scope;
                short dst_col = iter->first->type() * SCOPE_NUMBER + iter->second;
                if(scope_checking_tbl_[src_row][dst_col]) {
                  dup=msg->clone();
                  ACE_DEBUG ((LM_DEBUG, "msg [%s] delivered once\n", ID2STR(msg->type).c_str()));
                  iter->first->put_msg(dup, timeout);
                }
              }
            }
          }
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
          typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.find(msg->type[i]);
          if (iter == node->children.end()) {
            //ACE_DEBUG((LM_DEBUG, "TrieRouter::route_msg: not found [%s]\n",ID2STR(msg->type).c_str()));
            return FAILURE;
          }
          node = iter->second;
        }

        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->registry.subers_lock_, FAILURE);

        //dispatch wildcard subscribers
        reg = &(node->registry);
        if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
          for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
              iter != reg->w_subers_->end(); iter++) {
            short src_row = src_type * SCOPE_NUMBER + pub_scope;
            short dst_col = iter->first->type() * SCOPE_NUMBER + iter->second;
            if(scope_checking_tbl_[src_row][dst_col]) {
              dup=msg->clone();
              ACE_DEBUG ((LM_DEBUG, "msg [%s] delivered once\n", ID2STR(msg->type).c_str()));
              iter->first->put_msg(dup, timeout);
            }
          }
        }

        //dispatch normal subscribers
        reg = &(node->registry);
        if (reg->subers_ != NULL && !reg->subers_->empty()) {
          for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->begin(); 
              iter != reg->subers_->end(); iter++) {
            short src_row = src_type * SCOPE_NUMBER + pub_scope;
            short dst_col = iter->first->type() * SCOPE_NUMBER + iter->second;
            if(scope_checking_tbl_[src_row][dst_col]) {
              dup=msg->clone();
              ACE_DEBUG ((LM_DEBUG, "msg [%s] delivered once\n", ID2STR(msg->type).c_str()));
              iter->first->put_msg(dup, timeout);
            }
          }
        }

        delete msg;

        return SUCCESS;
      }


    Status subscribed_global_msgs(std::vector<IdType> &global_msgs)
      {
        IdType id;
        return subscribed_global_msgs_recursive(global_msgs, sub_trie_root_, id);
      }
    Status subscribed_global_msgs_recursive(std::vector<IdType> &global_msgs, TrieNode<Sub_Registry> *node, IdType &id)
      {
        Sub_Registry *reg = &(node->registry);

        {
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, reg->subers_lock_, FAILURE);

          //check wildcard subscription
          if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
            for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->w_subers_->begin();
                iter != reg->w_subers_->end(); iter++) 
              if (iter->first->type() == MEMBER_LOCAL && 
                  (iter->second == SCOPE_GLOBAL || iter->second == SCOPE_REMOTE)) {
                id.push_back(IdTrait::WildcardName);
                global_msgs.push_back(id);
                id.pop_back();
                return SUCCESS; //wildcard stop going deeper
              }
          }

          //check normal subscription
          if (reg->subers_ != NULL && !reg->subers_->empty()) {
            for(typename std::map<Destination *, PubSub_Scope>::iterator iter = reg->subers_->begin();
                iter != reg->subers_->end() ; iter++) 
              if (iter->first->type() == MEMBER_LOCAL && 
                  (iter->second == SCOPE_GLOBAL || iter->second == SCOPE_REMOTE)) {
                global_msgs.push_back(id);
                break;
              }
          }
        }

        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
        for(typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.begin();
            iter != node->children.end(); iter++)
          if (iter->second != NULL) {
            id.push_back(iter->first);
            subscribed_global_msgs_recursive(global_msgs, iter->second, id);
            id.pop_back();
          }

        return SUCCESS;
      }

    Status published_global_msgs(std::vector<IdType> &global_msgs)
      {
        IdType id;
        return published_global_msgs_recursive(global_msgs, pub_trie_root_, id);
      }
    Status published_global_msgs_recursive(std::vector<IdType> &global_msgs, TrieNode<Pub_Registry> *node, IdType &id)
      {
        Pub_Registry *reg = &(node->registry);

        {
          ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, reg->pubers_lock_, FAILURE);
          //check wildcard 
          if (reg->w_pubers_ != NULL && !reg->w_pubers_->empty()) {
            id.push_back(IdTrait::WildcardName);
            global_msgs.push_back(id);
            id.pop_back();
            return SUCCESS; //wildcard stop going deeper
          }
          //check normal subscription
          if (reg->pubers_ != NULL && !reg->pubers_->empty()) {
            global_msgs.push_back(id);
          }
        }

        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
        for(typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.begin();
            iter != node->children.end(); iter++)
          if (iter->second != NULL) {
            id.push_back(iter->first);
            published_global_msgs_recursive(global_msgs, iter->second, id);
            id.pop_back();
          }

        return SUCCESS;
      }

    void dump_pub_ids_recursive (TrieNode<Pub_Registry> *node, IdType &id) {
      Pub_Registry *reg = &(node->registry);

      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, reg->pubers_lock_, FAILURE);
        //check wildcard 
        if (reg->w_pubers_ != NULL && !reg->w_pubers_->empty()) {
          id.push_back(IdTrait::WildcardName);
          ACE_DEBUG((LM_DEBUG, "%s\n", ID2STR(id).c_str()));
          id.pop_back();
        }
        //check normal subscription
        if (reg->pubers_ != NULL && !reg->pubers_->empty()) {
          ACE_DEBUG((LM_DEBUG, "%s\n", ID2STR(id).c_str()));
        }
      }

      ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
      for(typename std::map<NameType, TrieNode<Pub_Registry> *>::iterator iter = node->children.begin();
          iter != node->children.end(); iter++)
        if (iter->second != NULL) {
          id.push_back(iter->first);
          dump_pub_ids_recursive(iter->second, id);
          id.pop_back();
        }
    }

    void dump_sub_ids_recursive (TrieNode<Sub_Registry> *node, IdType &id) {
      Sub_Registry *reg = &(node->registry);

      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, reg->subers_lock_, FAILURE);
        //check wildcard 
        if (reg->w_subers_ != NULL && !reg->w_subers_->empty()) {
          id.push_back(IdTrait::WildcardName);
          ACE_DEBUG((LM_DEBUG, "%s\n", ID2STR(id).c_str()));
          id.pop_back();
        }
        //check normal subscription
        if (reg->subers_ != NULL && !reg->subers_->empty()) {
          ACE_DEBUG((LM_DEBUG, "%s\n", ID2STR(id).c_str()));
        }
      }

      ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, node->children_lock_, FAILURE);
      for(typename std::map<NameType, TrieNode<Sub_Registry> *>::iterator iter = node->children.begin();
          iter != node->children.end(); iter++)
        if (iter->second != NULL) {
          id.push_back(iter->first);
          dump_sub_ids_recursive(iter->second, id);
          id.pop_back();
        }
    }

    void dump_routing_tables(void)
      {
        ACE_DEBUG((LM_DEBUG, "The following IdTypes subscribed: \n"));
        {
          IdType id;
          dump_sub_ids_recursive(sub_trie_root_,id);
        }

        ACE_DEBUG((LM_DEBUG, "The following IdTypes published: \n"));
        {
          IdType id;
          dump_pub_ids_recursive(pub_trie_root_,id);
        }
      }

  protected:
    Status propagate_change_to_neighbors(Oper_Type op, IdType t)
      {
        ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors [%s]\n",ID2STR(t).c_str()));
        PubSub_Info_Msg *sub = new PubSub_Info_Msg();
        sub->num_msg_types = 1;
        sub->msg_types[0] = t;
        IdType mt;
        switch(op) {
        case OPER_PUBLISH:
          mt = Channel::PUBLICATION_INFO_MSG;
          ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors publish "));
          break;
        case OPER_UNPUBLISH:
          mt = Channel::UNPUBLICATION_INFO_MSG;
          ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors unpublish "));
          break;
        case OPER_SUBSCRIBE:
          mt = Channel::SUBSCRIPTION_INFO_MSG;
          ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors subscribe "));
          break;
        case OPER_UNSUBSCRIBE:
          mt = Channel::UNSUBSCRIPTION_INFO_MSG;
          ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors unsubscribe "));
          break;
        default:
          ACE_DEBUG((LM_DEBUG, "TrieRouter:::propagate_change_to_neighbors invalid IdType "));
          delete sub;
          return SUCCESS;
        }
        Msg *m = new Msg(mt, sub);
        route_msg(m, MEMBER_LOCAL, SCOPE_REMOTE);
        return SUCCESS;
      }
  };

};



#endif

---