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

#include "ace/Message_Queue.h"
#include "ace/Log_Msg.h"

#include <BaseDef.h>


namespace channel {

  template <class Channel, class MsgQueueType=ACE_Message_Queue<ACE_MT_SYNCH> >
  class  Port : public Channel::Destination, public Channel::Source {
    typedef typename Channel::Msg Msg;
    
  private:
    Channel *my_chan_;
    MsgQueueType *my_que_;
    ACE_Thread_Semaphore suspend_sema_;
    int num_blocked_thr;
    bool borrow_que_;

  public:

    Port(Channel *chan, MsgQueueType *que=NULL) : 
      Channel::Destination(chan), Channel::Source(chan)
      {
        num_blocked_thr = 0;
        borrow_que_ = true;
        my_chan_ = chan;
        my_que_ = que;
        if (que == NULL) {
          //allocate a new private msg_que
          my_que_ = new MsgQueueType();
          borrow_que_ = false;
        }
      }

    ~Port() {
      if (!borrow_que_)
        delete my_que_;
    }

    Status put_msg(Msg *msg, ACE_Time_Value *timeout) {
      //Note: since here mb (msg block) doesn't create/own the mem space for msg,
      //mb and its data will be release() and delete separately and explicitly.
      //internally it is marked as DONT_DELETE
      ACE_Message_Block *mb = new ACE_Message_Block((const char*) msg, sizeof(Msg));
      mb->wr_ptr (sizeof(Msg));
      if(my_que_->enqueue_tail(mb, timeout)!=-1)
        return SUCCESS;
      return FAILURE;
    }

    int num_pending_msg(void) {
      return my_que_->message_count();
    }

    Status recv_msg(Msg *&msg)
      {
        ACE_Message_Block *mb = 0;
        int ret;

        if(my_que_->state()==MsgQueueType::PULSED) {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) task block on semaphore when que is pulsed\n"));
          num_blocked_thr++;
          ret = suspend_sema_.acquire();
        }

        if(my_que_->dequeue_head (mb) == -1) {  //wait indefinitely
          if(my_que_->state()!=MsgQueueType::PULSED)
            return FAILURE;     
        
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) task block on semaphore when que is pulsed\n"));
          num_blocked_thr++;
          ret = suspend_sema_.acquire();
        }
        //forward message
        msg = (Msg *)mb->rd_ptr();
        //dont worry about msg (the internal data), it is already
        //marked DONT_DELETE when msg is wrapped into mblk
        mb->release();
        return SUCCESS;
      }

    Status suspend(void)
      {
        my_que_->pulse();
        return SUCCESS; //add later
      }
    Status resume(void)
      {
        my_que_->activate();
        for(int i=0; i<num_blocked_thr;i++)
          suspend_sema_.release();  //release my threads
        return SUCCESS; //add later
      }

  };

};


#endif

---