L:/channel/channel/include/Member.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 _MEMBER_H_
#define _MEMBER_H_

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

//std headers
#include <map>

//export 
#include <BaseDef.h>

namespace channel {

  template <class Channel>
    class  Member {
    public:
    //cannot use the following constructor because it requires the
    //most derived class to construct virtual base class explicitly
    //Member(Channel *c) : ch_(c) {}
    Member() : ch_(NULL) {}
    virtual ~Member() {};
    virtual Member_Type type() { return MEMBER_LOCAL; } //default local members
    protected:
    Channel *ch_;
  };    

  template <class, int> class Source;
  template <class, int> class Destination;


  template <class Channel>
    class Source<Channel, NAMESPACE_LINEAR> : virtual public Member<Channel> {
    public:
    typedef typename Channel::IdType IdType;
    typedef typename Channel::Msg Msg;
    typedef typename Channel::SynchPolicy SynchPolicy;

    Source (Channel *c) { Member<Channel>::ch_ = c; }
    virtual ~Source() {};
    Status publish_msg(IdType t, PubSub_Scope s=SCOPE_GLOBAL)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] already published, must unpublish it before republish\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        msg_scope_[t] = s;
        return this->ch_->publish_msg(t,s,this);
      }
    Status unpublish_msg(IdType t)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] not published, cannot be unpublished\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        PubSub_Scope s = msg_scope_[t];
        msg_scope_.erase(t);
        return this->ch_->unpublish_msg(t,s,this);
      }
    Status unpublish_all(void)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter =  msg_scope_.begin();
            iter !=  msg_scope_.end(); iter++) {
          this->ch_->unpublish_msg(iter->first, iter->second, this);
        }
        msg_scope_.clear();
        return SUCCESS;
      }
    template <class MsgType>
    Status send_msg(IdType t, MsgType *msg, int sz = sizeof(MsgType), 
                            MsgFreeCallback g = DefaultMsgFreeCallback<MsgType>,
                            ACE_Time_Value *timeout=0) {
      Msg *m = new Msg(t, msg, sz, g);
      return send_msg (m, timeout);
    }
    virtual Status send_msg(Msg *msg, ACE_Time_Value *timeout=0)
      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if (msg_scope_.find(msg->type) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "src sends msg [%s]\n", ID2STR(msg->type).c_str()));
          return this->ch_->route_msg(msg, this->type(), msg_scope_[msg->type], timeout);
        } else {
          ACE_DEBUG((LM_DEBUG, "src failed to sends unpublished msg [%s], dropped\n", ID2STR(msg->type).c_str()));
          delete msg;
        }
        return FAILURE;
      }
    PubSub_Scope scope(IdType t)
      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, SCOPE_UNDEFINED);
        if(msg_scope_.find(t) != msg_scope_.end())
          return msg_scope_[t];
        else
          return SCOPE_UNDEFINED;
      }
    protected:
    //published msgs scopes
    std::map<IdType, PubSub_Scope> msg_scope_;
    typename SynchPolicy::RW_MUTEX scope_lock_;
  };

  template <class Channel>
    class Destination<Channel, NAMESPACE_LINEAR> : virtual public Member<Channel> {
    public:
    typedef typename Channel::IdType IdType;
    typedef typename Channel::Msg Msg;
    typedef typename Channel::SynchPolicy SynchPolicy;

    Destination (Channel *c) { Member<Channel>::ch_ = c; }
    virtual ~Destination() {}
    Status subscribe_msg(IdType t, PubSub_Scope s=SCOPE_GLOBAL)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] already subscribed, must unsubscribe it before resubscribe\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        msg_scope_[t] = s;
        return this->ch_->subscribe_msg(t, s, this);
      }
    Status unsubscribe_msg(IdType t)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] not subscribed, cannot be unsubscribed\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        msg_scope_.erase(t);
        return this->ch_->unsubscribe_msg(t, this);
      }
    Status unsubscribe_all(void)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter =  msg_scope_.begin();
            iter !=  msg_scope_.end(); iter++) {
          this->ch_->unsubscribe_msg(iter->first, this);
        }
        msg_scope_.clear();
        return SUCCESS;
      }
    virtual Status put_msg(Msg *msg, ACE_Time_Value *timeout=0) = 0;
    //utils
    bool operator==(Destination *s) { 
      return equals(s);
    }
    bool equals(Destination *s) { 
      if (this->type() != s->type()) return false;
      if (((Destination *)this) != s) return false;
      return true;
    }
    PubSub_Scope scope(IdType t)
      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, SCOPE_UNDEFINED);
        if(msg_scope_.find(t) != msg_scope_.end())
          return msg_scope_[t];
        else
          return SCOPE_UNDEFINED;
      }
    protected:
    //subscribed msgs scopes
    std::map<IdType, PubSub_Scope> msg_scope_;
    typename SynchPolicy::RW_MUTEX scope_lock_;
  }; 


  template <class Channel>
    class Source<Channel, NAMESPACE_HIERARCHICAL> : virtual public Member<Channel> {
    public:
    typedef typename Channel::IdType IdType;
    typedef typename Channel::IdTrait IdTrait;
    typedef typename Channel::Msg Msg;
    typedef typename Channel::SynchPolicy SynchPolicy;

    Source (Channel *c) { Member<Channel>::ch_ = c; }
    virtual ~Source() {};
    Status publish_msg(IdType t, PubSub_Scope s=SCOPE_GLOBAL)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter = msg_scope_.begin();
            iter != msg_scope_.end(); iter++)
          if (IdTrait::match(iter->first, t)) {
            ACE_DEBUG((LM_DEBUG, "Msg Id [%s] conflict, cannot be published\n", ID2STR(t).c_str()));
            return FAILURE;
          }
        msg_scope_[t] = s;
        return this->ch_->publish_msg(t,s,this);
      }
    Status unpublish_msg(IdType t)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] not published, cannot be unpublished\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        PubSub_Scope s = msg_scope_[t];
        msg_scope_.erase(t);
        return this->ch_->unpublish_msg(t,s,this);
      }
    Status unpublish_all(void)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter =  msg_scope_.begin();
            iter !=  msg_scope_.end(); iter++) {
          this->ch_->unpublish_msg(iter->first, iter->second, this);
        }
        msg_scope_.clear();
        return SUCCESS;
      }
    template <class MsgType>
    Status send_msg(IdType t, MsgType *msg, int sz = sizeof(MsgType), 
                            MsgFreeCallback g = DefaultMsgFreeCallback<MsgType>,
                            ACE_Time_Value *timeout=0) {
      Msg *m = new Msg(t, msg, sz, g);
      return send_msg (m, timeout);
    }
    virtual Status send_msg(Msg *msg, ACE_Time_Value *timeout=0)
      {
        //cannot use wildcard when sending
        if (IdTrait::endWithWildcard(msg->type)) {
          ACE_DEBUG((LM_DEBUG, "TrieRouter::route_msg: Wildcard in Id [%s]\n",ID2STR(msg->type).c_str()));
          delete msg;
          return FAILURE;
        }

        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter = msg_scope_.begin();
            iter != msg_scope_.end(); iter++)
          if (IdTrait::id1contains2(iter->first, msg->type)) {
            return this->ch_->route_msg(msg, this->type(), iter->second, timeout);
          }
        ACE_DEBUG((LM_DEBUG, "src failed to sends unpublished msg [%s], dropped\n", ID2STR(msg->type).c_str()));
        delete msg;
        return FAILURE;
      }
    PubSub_Scope scope(IdType t)
      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, SCOPE_UNDEFINED);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter = msg_scope_.begin();
            iter != msg_scope_.end(); iter++)
          if (IdTrait::id1contains2(iter->first, t)) {
            return iter->second;
          }
        else
          return SCOPE_UNDEFINED;
      }
    protected:
    //published msgs scopes
    std::map<IdType, PubSub_Scope> msg_scope_;
    typename SynchPolicy::RW_MUTEX scope_lock_;
  };

  template <class Channel>
    class Destination<Channel, NAMESPACE_HIERARCHICAL> : virtual public Member<Channel> {
    public:
    typedef typename Channel::IdType IdType;
    typedef typename Channel::IdTrait IdTrait;
    typedef typename Channel::Msg Msg;
    typedef typename Channel::SynchPolicy SynchPolicy;

    Destination (Channel *c) { Member<Channel>::ch_ = c; }
    virtual ~Destination() {}
    Status subscribe_msg(IdType t, PubSub_Scope s=SCOPE_GLOBAL)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter = msg_scope_.begin();
            iter != msg_scope_.end(); iter++)
          if (IdTrait::match(iter->first, t)) {
            ACE_DEBUG((LM_DEBUG, "Msg Id [%s] conflict, cannot be subscribed\n", ID2STR(t).c_str()));
            return FAILURE;
          }
        msg_scope_[t] = s;
        return this->ch_->subscribe_msg(t, s, this);
      }
    Status unsubscribe_msg(IdType t)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        if(msg_scope_.find(t) != msg_scope_.end()) {
          ACE_DEBUG((LM_DEBUG, "Msg Id [%s] not subscribed, cannot be unsubscribed\n", ID2STR(t).c_str()));
          return FAILURE;
        }
        msg_scope_.erase(t);
        return this->ch_->unsubscribe_msg(t, this);
      }
    Status unsubscribe_all(void)
      {
        ACE_WRITE_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, FAILURE);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter =  msg_scope_.begin();
            iter !=  msg_scope_.end(); iter++) {
          this->ch_->unsubscribe_msg(iter->first, this);
        }
        msg_scope_.clear();
        return SUCCESS;
      }
    virtual Status put_msg(Msg *msg, ACE_Time_Value *timeout=0) = 0;
    //utils
    bool operator==(Destination *s) { 
      return equals(s);
    }
    bool equals(Destination *s) { 
      if (this->type() != s->type()) return false;
      if (((Destination *)this) != s) return false;
      return true;
    }
    PubSub_Scope scope(IdType t)
      {
        ACE_READ_GUARD_RETURN(typename SynchPolicy::RW_MUTEX, guard, scope_lock_, SCOPE_UNDEFINED);
        for(typename std::map<IdType, PubSub_Scope>::iterator iter = msg_scope_.begin();
            iter != msg_scope_.end(); iter++)
          if (IdTrait::id1contains2(iter->first, t)) {
            return iter->second;
          }
        else
          return SCOPE_UNDEFINED;
      }
    protected:
    //subscribed msgs scopes
    std::map<IdType, PubSub_Scope> msg_scope_;
    typename SynchPolicy::RW_MUTEX scope_lock_;
  }; 

  template <class Channel>
    class  Callback : public Channel::Destination {
    public:
    typedef typename Channel::Msg Msg;

    Callback (Channel *c) : Channel::Destination(c) {}
    Status put_msg(Msg *msg, ACE_Time_Value *timeout) {
      ACE_UNUSED_ARG(timeout);
      return process (msg);
    }
    virtual Status process(Msg *msg) = 0;
  };


};

#endif

---