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

//ACE headers 
#include "ace/Thread_Mutex.h"
#include "ace/Thread_Semaphore.h"
#include "ace/Synch.h"
#include "ace/OS.h"
#include "ace/os_include/os_netdb.h"
#include "ace/Log_Msg.h"
#include "ace/Synch_Options.h"
#include "ace/Singleton.h"
#include "ace/Recursive_Thread_Mutex.h"
#include "ace/Thread_Manager.h"
#include "ace/Reactor.h"

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

//channel
#include <BaseDef.h>

namespace channel {
 
  //ask the main-loop of connector's reactor to shutdown nicely
  class Quit_Handler : public ACE_Event_Handler {
    friend class ace_dewarn_gplusplus;
  public:
    Quit_Handler (ACE_Reactor *r) : ACE_Event_Handler (r) {}

    virtual int handle_exception (ACE_HANDLE) {
      ACE_DEBUG((LM_DEBUG, "Channel Reactor shutting down now...\n"));
      reactor ()->end_reactor_event_loop ();
      return -1; // Trigger call to handle_close() method.
    }

    virtual int handle_close (ACE_HANDLE, ACE_Reactor_Mask)
      { delete this; return 0; }

  private:

    // Private destructor ensures dynamic allocation.
    virtual ~Quit_Handler () {}
  };

  template <class, class> class ConnHandler;
  template <class, class> class OutputMgr;
  class ConnInfo;
  template <class, class> class MarshalerRegistry;
  template <class, class> class BinderRegistry;
  template <class, class> class Binder;
  template <class, class> class Filter;
  template <class, class> class Translator;
  class Marshaler;

  template <class Channel, class Transport>
  class Connector : public Transport, 
    public MarshalerRegistry<typename Channel::IdType, typename Channel::IdTrait>,
    public BinderRegistry<typename Channel::IdType, typename Channel::IdTrait> {
    template <class, class> friend class ConnHandler;
    template <class, class> friend class RemoteInterface;

  public:

    typedef ConnHandler<Channel, Transport> ConnHandler;
    typedef Interface<Channel> Interface;
    typedef RemoteInterface<Channel, ConnHandler> RemoteInterface;
    typedef Binder<typename Channel::IdType, typename Channel::IdTrait> Binder;
    typedef Filter<typename Channel::IdType, typename Channel::IdTrait> Filter;
    typedef Translator<typename Channel::IdType, typename Channel::IdTrait> Translator;

    bool exit_start_; 
    ACE_Thread_Semaphore exit_sema_; 

  protected:

    OutputMgr<Channel, ConnHandler> output_mgr_;
    int num_thr_;
    bool driver_thread_;
    std::map<ConnInfo, ConnHandler *> conn_map_; 
    ACE_Thread_Mutex conn_map_lock_;

  public:

    Channel *ch_;

    Connector(Channel *mchan, bool dt = false, int nt=1) : Transport(this) {
      conn_map_.clear();
      exit_start_ = false;
      ch_ = mchan;
      num_thr_ = nt;
      driver_thread_ = dt;
    }
    ~Connector() {}

    static Interface_Type type(void) { 
      return Transport::type(); }

    //the main loop driving connector's reactor
    static void * run_event_loop(void *arg) {
      ACE_UNUSED_ARG(arg);
      ACE_DEBUG((LM_DEBUG, "A event_loop thread started\n"));
      ACE_Reactor *reactor = ACE_Reactor::instance();
      reactor->owner(ACE_OS::thr_self());
      reactor->run_reactor_event_loop();
      return 0;
    }

    Status open(std::string addr)
      {
        output_mgr_.num_thr(num_thr_);
        output_mgr_.open();

        ConnInfo ci(addr);
        Transport::open(ci);

        if (driver_thread_) startup();

        return SUCCESS;
      }

    Status open(void)
      {
        output_mgr_.num_thr(num_thr_);
        output_mgr_.open();

        if (driver_thread_) startup();

        return SUCCESS;
      }

    Status startup(void) {
      ACE_Thread_Manager::instance()->spawn(run_event_loop);
      return SUCCESS;
    }

    Status shutdown(void) {
      ACE_Reactor *reactor = ACE_Reactor::instance();
      Quit_Handler *quit_handler = 0;
      ACE_NEW_RETURN (quit_handler, Quit_Handler (reactor), 0);
      reactor->notify (quit_handler);
      return SUCCESS;
    }
  
    Status close(void)
      {
        ACE_DEBUG ((LM_DEBUG, "(%t) Connector::close() before lock conn_map_lock...\n"));
        ACE_GUARD_RETURN (ACE_Thread_Mutex, guard, conn_map_lock_, FAILURE);
        ACE_DEBUG ((LM_DEBUG, "(%t) Connector::close() after lock conn_map_lock...\n"));
        ACE_DEBUG ((LM_DEBUG, "(%t) Connector first shutdown all writer threads...\n"));
        output_mgr_.shut_down(); 
        ACE_DEBUG ((LM_DEBUG, "(%t) Connector shutdown all reader threads...\n"));
        exit_start_=true;
        typename std::map<ConnInfo, ConnHandler *>::iterator iter;
        for(iter=conn_map_.begin(); iter!=conn_map_.end(); iter++)
          iter->second->shut_down();
        for(iter=conn_map_.begin(); iter!=conn_map_.end(); iter++)
          exit_sema_.acquire();

        Transport::close();

        return SUCCESS;
      }

    Status connect(std::string addr_str, Binder *b = NULL)
      {
        ConnInfo addr(addr_str);
        return connect(addr, b);
      }

    Status connect(ConnInfo addr, Binder *bind = NULL)
      {
        ConnInfo adr2(addr);
        Transport::get_map_index(adr2);
        if(is_connected(adr2))
          return SUCCESS;
        ConnHandler *ch = NULL;
        if(Transport::connect (addr, ch) != SUCCESS)
          return FAILURE;
        ch->rmt_addr_ = adr2;
        //register binders(filters+translators) here
        if (bind != NULL)
          register_binder(adr2, bind);
        return SUCCESS;
      }

    Status disconnect(std::string addr) { 
      ACE_UNUSED_ARG(addr);
      return SUCCESS; }
    
    Status add_conn(ConnInfo peer_addr, ConnHandler *ch)
      {
        ACE_GUARD_RETURN (ACE_Thread_Mutex, guard, conn_map_lock_, FAILURE);
        if(conn_map_.find(peer_addr) != conn_map_.end()) {
          ACE_DEBUG ((LM_DEBUG, "duplicate connection...\n"));
          return FAILURE; 
        }
        conn_map_[peer_addr] = ch;

        return SUCCESS;
      }
    Status del_conn(ConnHandler *ch)
      {
        ACE_GUARD_RETURN (ACE_Thread_Mutex, guard, conn_map_lock_, FAILURE);
        typename std::map<ConnInfo, ConnHandler*>::iterator iter;
        for(iter = conn_map_.begin(); iter != conn_map_.end(); iter++) 
          if(iter->second == ch) {
            conn_map_.erase(iter->first);
            break;
          }
        return SUCCESS;
      }
    Status del_conn(ConnInfo addr)
      {
        ACE_GUARD_RETURN (ACE_Thread_Mutex, guard, conn_map_lock_, FAILURE);
        conn_map_.erase(addr);
        return SUCCESS;
      }
    bool is_connected(ConnInfo addr)
      {
        ACE_GUARD_RETURN (ACE_Thread_Mutex, guard, conn_map_lock_, FAILURE);
        if(conn_map_.find(addr) != conn_map_.end())
          return true;
        return false;
      }

  };


};

#endif

---