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

#include "ace/SOCK_Stream.h"
#include "ace/Message_Block.h"
#include "ace/CDR_Stream.h"
#include "ace/Log_Msg.h"

#include <BaseDef.h>

namespace channel {

  template <class, class> class Connector;
  template <class, class> class RemoteInterface;
  class ConnInfo;

  template <class Channel, class Transport>
    class ConnHandler {
    public:
    typedef typename Channel::IdType IdType;
    typedef typename Channel::IdTrait IdTrait;
    typedef typename Channel::Msg Msg;
    typedef Connector<Channel, Transport> Connector;
    typedef ConnHandler<Channel, Transport> ConnHandlerType;
    typedef RemoteInterface<Channel, Transport> RemoteInterface;

    Connector *cm_; //my Connector
    ConnInfo rmt_addr_; //my peer "published" addr
    RemoteInterface *intf_; //my local interface to channel

    ConnHandler ()
      : cm_(NULL), intf_(NULL) {}
    virtual ~ConnHandler () {}
    virtual ACE_SOCK_Stream &peer_stream(void) = 0; //return my peer, either unix_domain or tcp sockets
    virtual void shut_down(void) = 0;

    int set_up(Connector *c, Interface_Role r)
      {
        cm_ = c;
        intf_ = new RemoteInterface(this, r, c->ch_);
        if (r == ACTIVE_ROLE) {
          //active side will add itself to conn_map here
          //passive side have to wait for the 1st msg
          if (c->add_conn (rmt_addr_, this) == FAILURE)
            return -1;

          //notify local channel members that peer connected
          intf_->notify_connected (rmt_addr_);

          // active side will initiate the start-up handshaking process
          if(intf_->send2remote_chan_info(Transport::type())==FAILURE) return -1;
        }
        return 0;
      }

    int service (void)
      {
        ConnInfo addr;
        ACE_DEBUG ((LM_DEBUG, "ConnHandler thread start...\n"));
        Msg *msg = 0;
        for (;;)
          switch (recv_message (msg)) {
          case -1: 
          case 0: 
            ACE_DEBUG ((LM_DEBUG, "(%t) error with reading socket or remote close connection...\n"));
            ACE_DEBUG ((LM_DEBUG, "(%t) one reader thread exits...\n"));
            //cleanup old data
            if(!cm_->exit_start_) {
              addr = rmt_addr_;
              delete intf_;
              cm_->del_conn(this);
            } else {
              ACE_DEBUG ((LM_DEBUG, "before closeup...\n"));
            }
            return 0;   // Client closed connection.
            break;
          default: 
            //ACE_DEBUG ((LM_DEBUG, "recv a msg...\n"));
            if (handle_message(msg) == FAILURE)
              return -1;
          }
        /* NOTREACHED */
        return 0;
      }

    Status handle_message(Msg *msg)
      {
        return intf_->handle_message(msg);
      }

    int recv_message(Msg *&msg)
      {
        //ACE_DEBUG ((LM_DEBUG,
        //            "(%t) start recving a msg...\n"));

        ACE_Message_Block *hdr =
          new ACE_Message_Block (ACE_DEFAULT_CDR_BUFSIZE);
        // Align the Message Block for a CDR stream
        ACE_CDR::mb_align(hdr);

        if (peer_stream().recv_n (hdr->wr_ptr (), 8) == 8) {
          //ACE_DEBUG ((LM_DEBUG,
          //          "(%t) recving a msg...1.5\n"));
          hdr->wr_ptr (8);               // Reflect addition of 8 bytes

          // Create a CDR stream to parse the 8-byte header.
          ACE_InputCDR hdr_cdr (hdr);

          // Extract the byte-order and use helper methods to
          // disambiguate octet, booleans, and chars.
          ACE_CDR::Boolean byte_order;
          hdr_cdr >> ACE_InputCDR::to_boolean (byte_order);

          // Set the byte-order on the stream...
          hdr_cdr.reset_byte_order (byte_order);

          // Extract the length
          ACE_CDR::Long msg_len;
          hdr_cdr >> msg_len;

          hdr->release ();
          hdr = 0;

          //the mem space for msg is allocated here inside mblk,
          //it will be deleted by default when release(). If it
          //need to be forwarded to other threads, need mark DONT_DELETE
          ACE_Message_Block *mb = new ACE_Message_Block (msg_len);
          // Align the Message Block for a CDR stream
          ACE_CDR::mb_align(mb);
  
          if (peer_stream().recv_n (mb->wr_ptr (), msg_len) == msg_len) {
            mb->wr_ptr (msg_len);

            ACE_InputCDR cdr (mb);
            cdr.reset_byte_order (byte_order);
            
            IdType type;
            IdTrait::demarshal(cdr, type);
            Marshaler *mar = cm_->get_marshaler(type);
            msg = new Msg();
            msg->type = type;
            msg->demarshal(cdr, mar);

            mb->release();

            ACE_DEBUG((LM_DEBUG, "read something from sock...\n"));
          
            return msg_len;
          } else {
            mb->release();
            mb = 0;
          }
        }

        hdr->release ();
        hdr = 0;
  
        ACE_DEBUG ((LM_DEBUG,
                      "(%t) something wrong when reading socket...\n"));

        return -1;
      }

    int send_msg (Msg *msg)
      {
        Marshaler *mar = cm_->get_marshaler(msg->type);

        ACE_DEBUG((LM_DEBUG, "ConnHandler::send_msg [%s]\n",ID2STR(msg->type).c_str()));

        const size_t max_payload_size = 
          4 //type_len
          //+ IdTrait<typename Channel::IdType>::size(msg->type)
          + IdTrait::size(msg->type)
          + 4 //data_len
          + msg->size_
          + ACE_CDR::MAX_ALIGNMENT; // padding;
        
        ACE_OutputCDR payload (max_payload_size);
        ACE_DEBUG((LM_DEBUG, "before msg marshal\n"));
        msg->marshal(payload, mar);
        ACE_DEBUG((LM_DEBUG, "after msg marshal\n"));
        ACE_CDR::ULong length = payload.total_length ();

        iovec iov[2];
        int iov_size = 0;

        ACE_OutputCDR header(ACE_CDR::MAX_ALIGNMENT+8);
        header << ACE_OutputCDR::from_boolean(ACE_CDR_BYTE_ORDER);
        header << ACE_CDR::ULong(length);

        iov[0].iov_base = header.begin()->rd_ptr();
        iov[0].iov_len = 8;
        iov[1].iov_base = payload.begin ()->rd_ptr ();
        iov[1].iov_len  = length;
        iov_size = 2;
    
        //acquire lock first
        //ACE_GUARD_RETURN(ACE_Thread_Mutex, guard, conn->lock_, -1);
        ACE_GUARD_REACTION(ACE_Thread_Mutex, guard, intf_->lock_, ACE_DEBUG ((LM_DEBUG, "failed to lock out-sock\n")); delete msg; return -1);
  
        // Send header and payload efficiently using "gather-write".
        peer_stream().sendv_n (iov, iov_size);
        //note: the msg_block don't own this out_msg (since it is not created
        //by this mblk; we have to delete it explicitly 
        delete msg;

        ACE_DEBUG((LM_DEBUG, "ConnHandler::send_msg OUT 1 msg\n"));

        return 0;
      }
  };

};

#endif

---