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

//for unix domain socket
#include "ace/LSOCK_Connector.h"
#include "ace/LSOCK_Acceptor.h"
#include "ace/UNIX_Addr.h"
#include "ace/SOCK_Stream.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"


//channel
#include <BaseDef.h>

//std
#include <iostream>
#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 UnixSockTransport;

  char *unix_path_prefix = "/tmp/";
  char *UNDEFINED_UNIX_ADDR ="undefined";
  char *DEFAULT_UNIX_ADDR ="chan_def_addr";

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

        return activate (THR_NEW_LWP | THR_DETACHED);
      }
    virtual int close(u_long flag=0)
      {
        ACE_UNUSED_ARG(flag);
        ACE_DEBUG ((LM_DEBUG, "(%t) Unix_ConnHandler<Channel>::close\n"));
        
        if(this->cm_ != NULL && this->cm_->exit_start_) {
          ACE_DEBUG ((LM_DEBUG, "(%t) Unix_ConnHandler<Channel>::close: release exit_sema_\n"));
          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
      //peer().close();
      ACE_DEBUG((LM_DEBUG, "one unix domain socket close...\n"));
      ACE_OS::close(get_handle());
    }
    virtual ACE_SOCK_Stream &peer_stream(void) {
      return (ACE_SOCK_Stream &) ACE_Svc_Handler<ACE_LSOCK_STREAM, ACE_NULL_SYNCH>::peer();
    }
    virtual int svc (void) { return ConnHandler<Channel, UnixSockTransport<Channel> >::service(); }
  };

  template <class Channel>
  class Unix_Acceptor
    : public ACE_Acceptor<Unix_ConnHandler<Channel>, ACE_LSOCK_Acceptor> {
    public:
    typedef ACE_Acceptor<Unix_ConnHandler<Channel>, ACE_LSOCK_Acceptor>
      PARENT;

    Connector<Channel, UnixSockTransport<Channel> > *cm_;
    virtual int activate_svc_handler (Unix_ConnHandler<Channel> *sh)
      {
        ACE_DEBUG((LM_DEBUG, "unix acceptor recv one conn...\n"));
        //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 Unix_Connector
    : public ACE_Connector<Unix_ConnHandler<Channel>, ACE_LSOCK_Connector> {
    public:
    typedef ACE_Connector<Unix_ConnHandler<Channel>, ACE_LSOCK_Connector>
      PARENT;

    Connector<Channel, UnixSockTransport<Channel> > *cm_;
    virtual int activate_svc_handler (Unix_ConnHandler<Channel> *sh)
      {
        ACE_DEBUG((LM_DEBUG, "unix connector detects one conn...\n"));
        //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  UnixSockTransport {
    typedef ConnHandler<Channel, UnixSockTransport<Channel> > ConnHandler;
    typedef Connector<Channel, UnixSockTransport<Channel> > Connector;
  private:

    // address I am listening for new connections
    std::string unix_addr_;
    // Factory that passively wait for connections
    Unix_Acceptor<Channel> unix_acceptor_;
    // Factory that actively connects to remotes
    Unix_Connector<Channel> unix_connector_;

  public:
    
    UnixSockTransport(Connector *c) {
      unix_acceptor_.cm_ = c;
      unix_connector_.cm_ = c;
      unix_addr_ = UNDEFINED_UNIX_ADDR;
    }
    static Interface_Type type(void) { return UNIX_SOCK; }

    //helpers
    std::string unix_addr(void) { return unix_addr_; }
    
    Status get_map_index(ConnInfo &addr)
      {
        //reuse current ConnInfo as conn_map_ index
        ACE_UNUSED_ARG(addr);
        return SUCCESS;
      }

    Status open(ConnInfo ci)
      {
        unix_addr_ = ci.unix_addr();

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

        if(unix_addr_ != UNDEFINED_UNIX_ADDR) {
          std::string unix_path = unix_path_prefix+unix_addr_;
          ACE_OS::unlink (unix_path.c_str());
          ACE_UNIX_Addr unaddr(unix_path.c_str());
          if(unix_acceptor_.open(unaddr) == -1) {//start listening for incoming connections
            ACE_DEBUG ((LM_DEBUG, "fail to open unix sock [%s]...\n", unix_path.c_str()));
            return FAILURE;
          }
          ACE_DEBUG ((LM_DEBUG, "open unix sock [%s]...\n", unix_path.c_str()));
        }
        ACE_DEBUG ((LM_DEBUG, "=================================\n"));
        return SUCCESS;
      }
    Status close(void)
      {
        if(unix_addr_ != UNDEFINED_UNIX_ADDR) 
          unix_connector_.close(); //?
        return SUCCESS;
      }

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

        Unix_ConnHandler<Channel> *handler=NULL;
        std::string unix_path = unix_path_prefix+addr.unix_addr();
        ACE_UNIX_Addr unaddr(unix_path.c_str());
      
        handler = new Unix_ConnHandler<Channel>();
        if(unix_connector_.connect(handler, unaddr, ACE_Synch_Options::synch) == -1) //ask connector create conn handler
          if (errno != EWOULDBLOCK) {
            ACE_DEBUG ((LM_DEBUG, " !!! server %s is DOWN !!!\n", unaddr.get_path_name()));
            delete handler;
            return FAILURE; //don't fail, go on for others
          }
        conn_handler = handler;
        ACE_DEBUG ((LM_DEBUG, "UnixSockConnector<Channel>::connect...finish...\n"));
        return SUCCESS;
      }
    Status disconnect(ConnInfo addr) { 
      ACE_UNUSED_ARG(addr);
      return SUCCESS; }

  };

};

#endif

---