L:/channel/cvm/include/CvmBaseConnector.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 _CVM_BASE_CONNECTOR_H_
#define _CVM_BASE_CONNECTOR_H_

#include "ace/Dynamic_Service.h"

#include <Channel.h>
#include <CvmService.h>

#include <iostream>
#include <vector>
#include <string>


namespace cvm {

  template <class Channel, class Transport>
    class CvmBaseConnector : public CvmService {
    public:
    typedef channel::Connector<Channel, Transport> Connector;

    Connector *conn_;
    Connector *conn(void) { return conn_; }
    virtual int init (int argc, ACE_TCHAR *argv[])
      {
        //-------------- Channel init -----------------
        ACE_TCHAR tcp_port[MAXHOSTNAMELEN];
        ACE_TCHAR unix_addr[MAXHOSTNAMELEN];
        ACE_TCHAR rmt_addr[MAXHOSTNAMELEN];
        int num_thr=1;  //default one thread in out_mgr pool
        strcpy(unix_addr, channel::UNDEFINED_UNIX_ADDR);
        std::vector<std::string> rmt_addrs;
        ACE_TCHAR cn[128];
        cn[0]='\0';

        conn_ = NULL;

        bool my_port_defined = false;
        bool my_unix_defined = false;
        bool rmt_addr_defined = false;

        ACE_Get_Opt get_opt (argc, argv, ACE_TEXT ("p:u:r:t:c:"), 0);
        get_opt.long_option (ACE_TEXT ("listen_port"), 'p',
                             ACE_Get_Opt::ARG_REQUIRED);
        get_opt.long_option (ACE_TEXT ("unix_addr"), 'u',
                             ACE_Get_Opt::ARG_REQUIRED);
        get_opt.long_option (ACE_TEXT ("rmt_addr"), 'r',
                             ACE_Get_Opt::ARG_REQUIRED);
        get_opt.long_option (ACE_TEXT ("num_thr"), 't',
                             ACE_Get_Opt::ARG_REQUIRED);
        get_opt.long_option (ACE_TEXT ("chan"), 'c',
                             ACE_Get_Opt::ARG_REQUIRED);

        for (int c; (c = get_opt ()) != -1;)
          switch (c) {
          case 'u': // unix_addr this chan is listening at
            my_unix_defined = true;
            ACE_OS::strsncpy
              (unix_addr, get_opt.opt_arg (), MAXHOSTNAMELEN);
            break;
          case 'p': // tcp port this chan is listening at
            my_port_defined = true;
            ACE_OS::strsncpy
              (tcp_port, get_opt.opt_arg (), MAXHOSTNAMELEN);
            break;
          case 't': // how many threads should the sock sender pool be
            num_thr = ACE_static_cast
              (int, ACE_OS::atoi (get_opt.opt_arg ()));
            break;
          case 'r': // remote chan addr to connect with
            rmt_addr_defined = true;
            ACE_OS::strsncpy
              (rmt_addr, get_opt.opt_arg (), MAXHOSTNAMELEN);
            rmt_addrs.push_back(std::string(rmt_addr));
            break;
          case 'c': // channel name
            ACE_OS::strsncpy
              (cn, get_opt.opt_arg (), 128);
            break;
          }
  
        
        ACE_DEBUG((LM_DEBUG, "(%t) thr_num=%d\n", num_thr));
        
        if (!my_port_defined && Connector::type() == channel::INET_SOCK) {
          ACE_DEBUG((LM_DEBUG, "(%t) No info about tcp port , failed...\n"));
          return -1;
        } else
          ACE_DEBUG((LM_DEBUG, "(%t) port=%d\n",tcp_port));

        if (!my_unix_defined && Connector::type() == channel::UNIX_SOCK) {
          ACE_DEBUG((LM_DEBUG, "(%t) No info about unix domain socket, failed...\n"));
          return -1;
        } else
          ACE_DEBUG((LM_DEBUG, "(%t) unix_addr=%s\n", unix_addr));

        Channel *ch_ = CvmBaseChannel<Channel>::find_chan(cn);
        if (ch_ == NULL) {
          ACE_DEBUG ((LM_DEBUG,
                      "(%t) Connector not bound to any Channel ...\n"));
          return -1;
        }
  
        conn_ = new Connector(ch_, false/*no-internal-driver-thr*/, num_thr);
        if(Connector::type() == channel::INET_SOCK) {
          if (conn_->open (tcp_port) == channel::FAILURE) return -1;
        } else if(my_unix_defined) {
          if (conn_->open (unix_addr) == channel::FAILURE) return -1;
        }

        if (rmt_addr_defined) {
          std::vector<std::string>::iterator iter;
          for(iter=rmt_addrs.begin(); iter!=rmt_addrs.end(); iter++) {
            channel::ConnInfo ci;
            ci.parse((*iter));
            if(ci.valid()) {
              switch(ci.type()) {
              case channel::INET_SOCK:
                if (conn_ != NULL && Connector::type() == channel::INET_SOCK) {
                  ACE_DEBUG((LM_DEBUG, "(%t) try to connect to ip=%s port=%d\n", ci.ip().c_str(), ci.port()));
                  if (conn_->connect (ci) == channel::FAILURE)
                    { 
                      continue;
                    }
                } else {
                  ACE_DEBUG((LM_DEBUG, "(%t) No tcp_connector drop ip=%s port=%d\n", ci.ip().c_str(), ci.port()));
                }
                break;
              case channel::UNIX_SOCK:
                if (conn_ != NULL && Connector::type() == channel::UNIX_SOCK) {
                  ACE_DEBUG((LM_DEBUG, "(%t) try to connect to unix_addr=%s\n" ,ci.unix_addr().c_str()));
                  if (conn_->connect (ci) == channel::FAILURE)
                    { 
                      continue;
                    }
                } else {
                  ACE_DEBUG((LM_DEBUG, "(%t) No unix_connector drop unix_addr=%s\n" ,ci.unix_addr().c_str())); 
                }
                break;
              default:
                break;
              }
            }
          }
        }

        ACE_DEBUG((LM_DEBUG, "### Connector is up and running ... ###\n"));

        return 0;
      }
    virtual int close (u_long)
      {
        //do nothing
        return 0;
      }
    virtual int fini ()
      {
        ACE_DEBUG((LM_DEBUG, "### Channel [%s] is shuting down ... ###\n", name().c_str()));

        conn_->close ();  
        delete conn_;

        ACE_DEBUG((LM_DEBUG, "Channel/Connector shutdown and ask all reader/writer threads exit...\n"));
        ACE_DEBUG((LM_DEBUG, "all reader/writeer threads exit...\n"));
        ACE_DEBUG((LM_DEBUG, "shutdown clock thread now...\n"));
        channel::Clock::instance()->stop();
        ACE_DEBUG((LM_DEBUG, "clock thread exit...\n"));
        //ACE_Reactor::end_event_loop ();
        ACE_OS::sleep(1);  //sleep 1 sec to give other threads another chance
        return 0;
      }
    virtual int info (ACE_TCHAR **strp, size_t length = 0) const
      {
        char buf[256];

        ACE_OS::sprintf (buf,
                         "CvmBaseConnector coming up ... \n");

        if (*strp == 0 && (*strp = ACE_OS::strdup (buf)) == 0)
          return -1;
        else
          ACE_OS::strncpy (*strp, buf, length);
        return ACE_OS::strlen (buf);
      }
    //virtual int suspend ();
    //virtual int resume ();

    static Connector * find_connector(const char* name) //find dynamic connectors
      {
        CvmBaseConnector<Channel, Transport> *d = ACE_Dynamic_Service<CvmBaseConnector<Channel, Transport> >::instance(ACE_TEXT(name));
        if (d==NULL)
          return NULL;
        else 
          return d->conn();
      }
  };

};

#endif

---