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

//for tcp socket
#include "ace/SOCK_Acceptor.h"
#include "ace/SOCK_Connector.h"
#include "ace/SOCK_Stream.h"
#include "ace/INET_Addr.h"
//acceptor-connector-svc_handler framework
#include "ace/Acceptor.h"
#include "ace/Connector.h"
#include "ace/Svc_Handler.h"
//for msg marshalling/demarshalling
#include "ace/CDR_Stream.h"
#include "ace/Message_Block.h"
//other major headers
#include "ace/Task.h"
#include "ace/Thread_Manager.h"
#include "ace/Thread_Mutex.h"
#include "ace/Thread_Semaphore.h"
#include "ace/Get_Opt.h"
#include "ace/Reactor.h"
#include "ace/Service_Object.h"
#include "ace/Signal.h"
#include "ace/Handle_Set.h"
#include "ace/FILE_Addr.h"
#include "ace/OS.h"
#include "ace/os_include/os_netdb.h"
#include "ace/Log_Msg.h"
#include "ace/Synch_Options.h"

//chan
#include <BaseDef.h>

//std
#include <iostream>
#include <sstream>
#include <string>
#include <sstream>
#include <vector>
#include <algorithm>
#include <netinet/in.h>

namespace channel {

  class ConnInfo;
  template <class, class> class ConnHandler;
  template <class, class> class Connector;
  template <class> class TcpSockTransport;

  u_short UNDEFINED_TCP_PORT=0;
  u_short DEFAULT_TCP_PORT=3456;

  template <class Channel>
  class Tcp_ConnHandler
    : public ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>, 
      public ConnHandler<Channel, TcpSockTransport<Channel> > {
    public:
    typedef ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH> PARENT;
    Tcp_ConnHandler () {}
    virtual int open (void *) // Initialization hook method.
      {
        ACE_DEBUG ((LM_DEBUG, "a new tcp conn comes up...\n"));

        int bufsiz = ACE_DEFAULT_MAX_SOCKET_BUFSIZ;
        peer ().set_option (SOL_SOCKET, SO_SNDBUF,
                            &bufsiz, sizeof bufsiz);

        return activate (THR_NEW_LWP | THR_DETACHED);
      }
    virtual int close(u_long flag=0)
      {
        ACE_UNUSED_ARG(flag);
        ACE_DEBUG ((LM_DEBUG, "(%t) Tcp_ConnHandler<Channel>::close\n"));
        if(this->cm_ != NULL && this->cm_->exit_start_) {
          ACE_DEBUG ((LM_DEBUG, "(%t) Tcp_ConnHandler<Channel>::close: release exit_sema_\n"));
          //Q: why need this->
          this->cm_->exit_sema_.release();
          //free resources
          PARENT::close();
        }
        return 0;
      }
    virtual void shut_down(void) {
      //close the stream, hope the reader thread will wake up
      //try to shutdown the thread and clean up data gracely
      ACE_DEBUG((LM_DEBUG,"one tcp socket close...\n"));
      ACE_OS::closesocket(get_handle());
      //peer().close();
    }
    virtual ACE_SOCK_Stream &peer_stream(void) {
      return (ACE_SOCK_Stream &) ACE_Svc_Handler<ACE_SOCK_STREAM, ACE_NULL_SYNCH>::peer();
    }
    virtual int svc (void) 
      { 
        return ConnHandler<Channel, TcpSockTransport<Channel> >::service(); 
      }
  };

  template <class Channel>
  class Tcp_Acceptor
    : public ACE_Acceptor<Tcp_ConnHandler<Channel>, ACE_SOCK_Acceptor> {
    public:
    typedef ACE_Acceptor<Tcp_ConnHandler<Channel>, ACE_SOCK_Acceptor>
      PARENT;

    Connector<Channel, TcpSockTransport<Channel> > *cm_;
    virtual int activate_svc_handler (Tcp_ConnHandler<Channel> *sh)
      {
        //hook up connection handler with framework
        sh->set_up (cm_, PASSIVE_ROLE);
        //activate threads etc
        if(PARENT::activate_svc_handler(sh)==-1) return -1;
        return 0;
      }

    //all use default behaviour
  };

  template <class Channel>
  class Tcp_Connector
    : public ACE_Connector<Tcp_ConnHandler<Channel>, ACE_SOCK_Connector> {
    public:
    typedef ACE_Connector<Tcp_ConnHandler<Channel>, ACE_SOCK_Connector>
      PARENT;

    Connector<Channel, TcpSockTransport<Channel> > *cm_;
    virtual int activate_svc_handler (Tcp_ConnHandler<Channel> *sh)
      {
        //hook up connection handler with framework
        sh->set_up (cm_, ACTIVE_ROLE);
        //activate threads etc
        if(PARENT::activate_svc_handler(sh)==-1) return -1;
        return 0;
      }

   //all use default behaviours 
  };

  template <class Channel>
    class  TcpSockTransport {
    typedef ConnHandler<Channel, TcpSockTransport<Channel> > ConnHandler;
    typedef Connector<Channel, TcpSockTransport<Channel> > Connector;
    private:

    // address I am listening for new connections
    std::string host_addr_;
    u_short port_;
    // Factory that passively wait for connections
    Tcp_Acceptor<Channel> tcp_acceptor_;
    // Factory that actively connects to remotes
    Tcp_Connector<Channel> tcp_connector_;

    public:

    typedef TcpSockTransport<Channel> TransportType;

    TcpSockTransport(Connector *c) {
      port_ = UNDEFINED_TCP_PORT;
      //pass cm to conn_handlers thru acceptor/connector
      tcp_acceptor_.cm_ = c;
      tcp_connector_.cm_ = c;
    }
    static Interface_Type type(void) { return INET_SOCK; }

    //helpers
    std::string host_addr(void) { return host_addr_; }
    int port(void) { return port_;}

    Status get_my_ip(std::string &addr) {
      struct hostent *host_info;
      ACE_utsname host_data;
      if (ACE_OS::uname (&host_data) < 0)
        return FAILURE;
      if ((host_info = ACE_OS::gethostbyname (ACE_TEXT_ALWAYS_CHAR(host_data.nodename))) == NULL)
        return FAILURE;
      long host,host_addr;
      ACE_OS::memcpy ((char *) &host,
                      (char *) host_info->h_addr,
                      host_info->h_length);
      host_addr = ntohl(host);
      ACE_DEBUG ((LM_DEBUG, "my name =[%s]  my addr=[%ld]\n", host_data.nodename, host_addr));
      ACE_INET_Addr inaddr(1234,host_addr);
      addr = inaddr.get_host_addr();
      ACE_DEBUG ((LM_DEBUG, "my_addr=[%s]\n",addr.c_str()));
      return SUCCESS;
    }

    Status get_map_index(ConnInfo &addr)
      {
        std::string addr_index;
        ACE_INET_Addr inaddr(addr.port(), addr.ip().c_str());
        if(std::string(addr.get_host_addr()).find(ACE_LOCALHOST)!=std::string::npos) {
          if (get_my_ip(addr_index) != SUCCESS)
            return FAILURE;
        }
        else
          addr_index = inaddr.get_host_addr();
        addr.set_host_addr (addr_index.c_str());
        return SUCCESS;
      }
    
    Status open(ConnInfo ci)
      {
        port_ = ci.port();

        ACE_DEBUG ((LM_DEBUG, "============= My Info ==============\n"));

        if(port_ != UNDEFINED_TCP_PORT) {
          ACE_INET_Addr addr(port_, INADDR_ANY);
          ACE_DEBUG ((LM_DEBUG, "port=%d host_name=%s\n",port_, addr.get_host_name()));

          if (get_my_ip(host_addr_) != SUCCESS)
            return FAILURE;
 
          if(tcp_acceptor_.open(addr) == -1) {//start listening for incoming connections
            ACE_DEBUG ((LM_DEBUG, "fail to open tcp lsitening sock...\n"));
            return FAILURE;
          }
        }
        ACE_DEBUG ((LM_DEBUG, "=================================\n"));
        return SUCCESS;
      }

    Status close(void)
      {
        if(port_ != UNDEFINED_TCP_PORT)
          tcp_acceptor_.close();  //?
        return SUCCESS;
      }

    Status connect(ConnInfo addr, ConnHandler *&conn_handler)
      {
        ACE_DEBUG ((LM_DEBUG, "TcpSockTransport<Channel>::connect...\n"));

        addr.dump();
        ACE_INET_Addr inaddr(addr.port(), addr.ip().c_str());

        Tcp_ConnHandler<Channel> *handler=NULL;
        handler = new Tcp_ConnHandler<Channel>();
        if(tcp_connector_.connect(handler, inaddr, ACE_Synch_Options::synch) == -1) //ask connector create conn handler
          if (errno != EWOULDBLOCK) {
            ACE_DEBUG ((LM_DEBUG, " !!! server %s | %d is DOWN !!!\n", inaddr.get_host_addr(), inaddr.get_port_number()));
            delete handler;
            return FAILURE; 
          }
        conn_handler = handler;
        ACE_DEBUG ((LM_DEBUG, "TcpSockTransport<Channel>::connect...finish...\n"));
        return SUCCESS;
      }
    Status disconnect(ConnInfo addr) { 
      ACE_UNUSED_ARG(addr);
      return SUCCESS; }

  };

};

#endif

---