ecg_mcast_gateway.cpp

这是广泛使用的通信开源项目,对于大容量,高并发的通讯要求完全能够胜任,他广泛可用于网络游戏医学图像网关的高qos要求.更详细的内容可阅读相应的材料

// ECG_Mcast_Gateway.cpp,v 1.7 2003/11/22 03:14:24 parsons Exp

#include "ECG_Mcast_Gateway.h"

#include "EC_Lifetime_Utils_T.h"
#include "ECG_Simple_Address_Server.h"
#include "ECG_Complex_Address_Server.h"
#include "ECG_Simple_Mcast_EH.h"
#include "ECG_Mcast_EH.h"
#include "ECG_UDP_EH.h"

#include "orbsvcs/orbsvcs/Event_Utilities.h"

#include "ace/Dynamic_Service.h"
#include "ace/Arg_Shifter.h"
#include "tao/ORB_Core.h"
#include "ace/OS_NS_strings.h"

#if ! defined (__ACE_INLINE__)
#include "ECG_Mcast_Gateway.i"
#endif /* __ACE_INLINE__ */

ACE_RCSID(Event, ECG_Mcast_Gateway, "ECG_Mcast_Gateway.cpp,v 1.7 2003/11/22 03:14:24 parsons Exp")

typedef TAO_EC_Shutdown_Command<TAO_EC_Servant_Var<TAO_ECG_UDP_Sender> >
UDP_Sender_Shutdown;

typedef TAO_EC_Shutdown_Command<TAO_EC_Servant_Var<TAO_ECG_UDP_Receiver> >
UDP_Receiver_Shutdown;


int
TAO_ECG_Mcast_Gateway::init_svcs (void)
{
  return ACE_Service_Config::static_svcs ()->
    insert (&ace_svc_desc_TAO_ECG_Mcast_Gateway);
}


int
TAO_ECG_Mcast_Gateway::fini (void)
{
  return 0;
}

int
TAO_ECG_Mcast_Gateway::init (int argc, char* argv[])
{
  int result = 0;

  ACE_Arg_Shifter arg_shifter (argc, argv);

  while (arg_shifter.is_anything_left ())
    {
      const char *arg = arg_shifter.get_current ();

      if (ACE_OS::strcasecmp (arg, "-ECGService") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const char* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, "receiver") == 0)
                this->service_type_ = ECG_MCAST_RECEIVER;
              else if (ACE_OS::strcasecmp (opt, "sender") == 0)
                this->service_type_ = ECG_MCAST_SENDER;
              else if (ACE_OS::strcasecmp (opt, "two_way") == 0)
                this->service_type_ = ECG_MCAST_TWO_WAY;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                                         "Unsupported <-ECGService> option "
                                         "value: <%s>. Ignoring this option "
                                         "- using defaults instead.",
                            opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGAddressServer") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const char* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, "basic") == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_BASIC;
              else if (ACE_OS::strcasecmp (opt, "source") == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_SOURCE;
              else if (ACE_OS::strcasecmp (opt, "type") == 0)
                this->address_server_type_ = ECG_ADDRESS_SERVER_TYPE;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                                  "Unsupported <-ECGAddressServer> "
                                  "option value: <%s>. Ignoring this "
                                  "option - using defaults instead.",
                              opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGAddressServerArg") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->address_server_arg_.set (arg_shifter.get_current ());
              arg_shifter.consume_arg ();
            }
        }


      else if (ACE_OS::strcasecmp (arg, "-ECGHandler") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const char* opt = arg_shifter.get_current ();
              if (ACE_OS::strcasecmp (opt, "basic") == 0)
                this->handler_type_ = ECG_HANDLER_BASIC;
              else if (ACE_OS::strcasecmp (opt, "complex") == 0)
                this->handler_type_ = ECG_HANDLER_COMPLEX;
              else if (ACE_OS::strcasecmp (opt, "udp") == 0)
                this->handler_type_ = ECG_HANDLER_UDP;
              else
                {
                  ACE_ERROR ((LM_ERROR,
                                  "Unsupported <-ECGHandler> "
                                  "option value: <%s>. Ignoring this "
                                  "option - using defaults instead.",
                              opt));
                  result = -1;
                }
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGTTL") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              const char* opt = arg_shifter.get_current ();
              unsigned long tmp = ACE_OS::strtoul (opt, 0, 0) & 0xff;
              this->ttl_value_ = ACE_static_cast (u_char, tmp);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGNIC") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->nic_.set (arg_shifter.get_current ());
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGIPMULTICASTLOOP") == 0)
        {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->ip_multicast_loop_ =
                (ACE_OS::atoi(arg_shifter.get_current()) != 0);
              arg_shifter.consume_arg ();
            }
        }

      else if (ACE_OS::strcasecmp (arg, "-ECGNONBLOCKING") == 0)
      {
          arg_shifter.consume_arg ();

          if (arg_shifter.is_parameter_next ())
            {
              this->non_blocking_ =
                (ACE_OS::atoi(arg_shifter.get_current()) != 0);
              arg_shifter.consume_arg ();
            }
        }

      else
        {
          arg_shifter.ignore_arg ();
          ACE_DEBUG ((LM_WARNING,
                             "Ignoring <%s> option "
                             "during initialization.",
                      arg));
          result = -1;
        }
    }

  if (this->validate_configuration () == -1)
    return -1;
  else
    return result;
}

int
TAO_ECG_Mcast_Gateway::init (const char * address_server_arg,
                             const Attributes & attr)
{
  this->address_server_arg_.set (address_server_arg);

  this->address_server_type_ = attr.address_server_type;
  this->handler_type_ = attr.handler_type;
  this->service_type_ = attr.service_type;
  this->ttl_value_ = attr.ttl_value;
  this->nic_.set (attr.nic.c_str ());
  this->ip_multicast_loop_ = attr.ip_multicast_loop;
  this->non_blocking_ = attr.non_blocking;

  return this->validate_configuration ();
}

int
TAO_ECG_Mcast_Gateway::init (
    const RtecEventChannelAdmin::ConsumerQOS & consumer_qos,
    const char * address_server_arg,
    const Attributes & attributes)
{
  this->consumer_qos_ = consumer_qos;
  return this->init (address_server_arg,
                     attributes);
}

int
TAO_ECG_Mcast_Gateway::validate_configuration (void)
{
  if ((this->handler_type_ == ECG_HANDLER_BASIC
       || this->handler_type_ == ECG_HANDLER_UDP)
      && this->service_type_ != ECG_MCAST_SENDER
      && this->address_server_type_ != ECG_ADDRESS_SERVER_BASIC)
    {
      ACE_DEBUG ((LM_ERROR,
                      "Configurations for mcast handler and "
                      "address server do not match."));
      return -1;
    }

  // Currently all Address Server implementations require an
  // initialization string.  If we ever add a new Address Server
  // implementation, which does not, we'll have to remove this check.
  if (this->address_server_arg_.length () == 0)
    {
      ACE_DEBUG ((LM_ERROR,
                      "Address server initializaton "
                      "argument not specified."));
      return -1;
    }

  if (this->ip_multicast_loop_ != 0
      && this->ip_multicast_loop_ != 1)
    {
      ACE_DEBUG ((LM_ERROR,
                  "IP MULTICAST LOOP option must have a boolean value."));
      return -1;
    }

  if (this->non_blocking_ != 0
      && this->non_blocking_ != 1)
    {
      ACE_DEBUG ((LM_ERROR,
                  "NON BLOCKING flag must have a boolean value."));
      return -1;
    }

  return 0;
}

TAO_ECG_Refcounted_Endpoint
TAO_ECG_Mcast_Gateway::init_endpoint (void)
{
  TAO_ECG_UDP_Out_Endpoint* endpoint = 0;
  TAO_ECG_Refcounted_Endpoint refendpoint;

  // Try to allocate a new endpoint from the heap
  ACE_NEW_NORETURN (endpoint,
                    TAO_ECG_UDP_Out_Endpoint);

  if (endpoint != 0)
  {
    refendpoint = endpoint;
  }
  else
  {
    return TAO_ECG_Refcounted_Endpoint ();
  }

  ACE_SOCK_Dgram& dgram = refendpoint->dgram ();

  if (dgram.open (ACE_Addr::sap_any) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                             "Cannot open dgram "
                             "for sending mcast messages."));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  if (this->nic_.length () != 0)
    {
      dgram.set_nic (this->nic_.c_str ());
    }

  if (this->ttl_value_ > 0)
    {
      if (dgram.ACE_SOCK::set_option (IPPROTO_IP,
                                      IP_MULTICAST_TTL,
                                      &this->ttl_value_,
                                      sizeof (this->ttl_value_))
          == -1)
        {
          ACE_ERROR ((LM_ERROR,
                      "Error setting TTL option on dgram "
                      "for sending mcast messages."));
          return TAO_ECG_Refcounted_Endpoint ();
        }
    }

  if (dgram.ACE_SOCK::set_option (IPPROTO_IP,
                                  IP_MULTICAST_LOOP,
                                  &this->ip_multicast_loop_,
                                  sizeof (this->ip_multicast_loop_)) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  "Error setting MULTICAST_LOOP option "
                  "on dgram for sending mcast messages."));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  if (this->non_blocking_
      && dgram.enable(ACE_NONBLOCK) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  "Error setting NON BLOCKING option."));
      return TAO_ECG_Refcounted_Endpoint ();
    }

  return refendpoint;
}

PortableServer::ServantBase *
TAO_ECG_Mcast_Gateway::init_address_server (void)
{
  const char * address_server_arg =
    (this->address_server_arg_.length ())