av_core.cpp

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

// AV_Core.cpp,v 5.22 2003/10/18 14:06:04 yamuna Exp

#include "orbsvcs/AV/AV_Core.h"
#include "orbsvcs/AV/FlowSpec_Entry.h"
#include "orbsvcs/AV/Transport.h"
#include "orbsvcs/AV/Protocol_Factory.h"
#include "orbsvcs/AV/UDP.h"
#include "orbsvcs/AV/TCP.h"
#include "orbsvcs/AV/RTP.h"
#include "orbsvcs/AV/RTCP.h"
#include "orbsvcs/AV/sfp.h"
#include "orbsvcs/AV/default_resource.h"

#if defined (ACE_HAS_RAPI) || defined (ACE_HAS_WINSOCK2_GQOS)
#include "orbsvcs/AV/QoS_UDP.h"
#endif /* ACE_HAS_RAPI || ACE_HAS_WINSOCK2_GQOS */

#if defined (ACE_HAS_SCTP)
#include "orbsvcs/AV/SCTP_SEQ.h"
#endif // ACE_HAS_SCTP

#include "tao/debug.h"
#include "tao/ORB_Core.h"

#include "ace/Dynamic_Service.h"

//------------------------------------------------------------
// TAO_AV_Core
//------------------------------------------------------------

TAO_AV_Core::TAO_AV_Core (void)
  :connector_registry_ (0),
   acceptor_registry_ (0)
{
  ACE_NEW (this->connector_registry_,
           TAO_AV_Connector_Registry
           );
  ACE_NEW (this->acceptor_registry_,
           TAO_AV_Acceptor_Registry
           );
}

TAO_AV_Core::~TAO_AV_Core (void)
{
  delete this->connector_registry_;
  delete this->acceptor_registry_;

  TAO_AV_TransportFactorySetItor transport_iter =
      this->transport_factories_.begin();

  while (transport_iter != this->transport_factories_.end())
    {
      if ((*transport_iter)->factory()->ref_count != 1)
        {
          delete (*transport_iter)->factory();
        }
      delete (*transport_iter);
      transport_iter++;
    }

  TAO_AV_Flow_ProtocolFactorySetItor flow_iter =
      this->flow_protocol_factories_.begin();

  while (flow_iter != this->flow_protocol_factories_.end())
    {
      if ((*flow_iter)->factory()->ref_count != 1)
        {
          delete (*flow_iter)->factory();
        }
      delete (*flow_iter);

      flow_iter++;
    }
}

CORBA::ORB_ptr
TAO_AV_Core::orb (void)
{
  return this->orb_.in ();
}

void
TAO_AV_Core::orb (CORBA::ORB_ptr orb)
{
  this->orb_ = orb;
}

PortableServer::POA_ptr
TAO_AV_Core::poa (void)
{
  return this->poa_.in ();
}

void
TAO_AV_Core::poa (PortableServer::POA_ptr poa)
{
  this->poa_ = poa;
}

TAO_AV_Connector_Registry*
TAO_AV_Core::connector_registry (void)
{
  return this->connector_registry_;
}

TAO_AV_Acceptor_Registry*
TAO_AV_Core::acceptor_registry (void)
{
  return this->acceptor_registry_;
}

TAO_AV_TransportFactorySet *
TAO_AV_Core::transport_factories (void)
{
  return &this->transport_factories_;
}

TAO_AV_Flow_ProtocolFactorySet*
TAO_AV_Core::flow_protocol_factories (void)
{
  return &this->flow_protocol_factories_;
}

int
TAO_AV_Core::stop_run (void)
{
  this->stop_run_ = 1;
  return 0;
}

int
TAO_AV_Core::run (void)
{
  this->stop_run_ = 0;
  while (!this->stop_run_ && this->orb_->work_pending ())
    this->orb_->perform_work ();
  return 0;
}

void
TAO_AV_Core::reactor (ACE_Reactor *r)
{
  this->reactor_ = r;
}

ACE_Reactor *
TAO_AV_Core::reactor (void)
{
  return this->reactor_;
}


int
TAO_AV_Core::init (CORBA::ORB_ptr orb,
                   PortableServer::POA_ptr poa
                   ACE_ENV_ARG_DECL_NOT_USED)
{
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_Core::init "));
  this->orb_ = CORBA::ORB::_duplicate (orb);
  this->poa_ = PortableServer::POA::_duplicate (poa);
  this->reactor (this->orb_->orb_core ()->reactor ());
  this->init_transport_factories ();
  this->init_flow_protocol_factories ();
  return 0;
}

int
TAO_AV_Core::init_forward_flows (TAO_Base_StreamEndPoint *endpoint,
                                 TAO_AV_FlowSpecSet &flow_spec_set,
                                 TAO_AV_Core::EndPoint direction,
                                 AVStreams::flowSpec &flow_spec)
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG,
                "TAO_AV_Core::init_forward_flows\n"));

  TAO_AV_FlowSpecSet address_flow_set;
  TAO_AV_FlowSpecSet flow_set;
  TAO_AV_FlowSpecSetItor end = flow_spec_set.end ();
  for (TAO_AV_FlowSpecSetItor start = flow_spec_set.begin ();
       start != end; ++start)
    {
      TAO_FlowSpec_Entry *entry = (*start);
      switch (direction)
        {
        case TAO_AV_Core::TAO_AV_ENDPOINT_B:
          {
            switch (entry->direction ())
              {
              case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                {
                  entry->role (TAO_FlowSpec_Entry::TAO_AV_CONSUMER);
                  break;
                }
              case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                {
                  entry->role (TAO_FlowSpec_Entry::TAO_AV_PRODUCER);
                  break;
                }
              }
            break;
          }
        case TAO_AV_Core::TAO_AV_ENDPOINT_A:
          {
            switch (entry->direction ())
              {
              case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                entry->role (TAO_FlowSpec_Entry::TAO_AV_PRODUCER);
                break;
              case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                entry->role (TAO_FlowSpec_Entry::TAO_AV_CONSUMER);
                break;
              }
            break;
          }
        default:
          break;
        }
      ACE_Addr *address = entry->address ();
      if (address != 0)
        {
          if (TAO_debug_level > 0)
            ACE_DEBUG ((LM_DEBUG,
                        "address given for flow %s\n",
                        entry->flowname ()));

          address_flow_set.insert (entry);
        }
      else
        flow_set.insert (entry);
    } //End of For Loop


  int result = -1;
  switch (direction)
    {
    case TAO_AV_Core::TAO_AV_ENDPOINT_A:
      if (address_flow_set.size () > 0)
        {
          result = this->acceptor_registry_->open (endpoint,
                                                   this,
                                                   address_flow_set);
          if (result < 0)
            ACE_ERROR_RETURN ((LM_ERROR,
                               "TAO_AV_Core::init_forward_flows::acceptor_registry::open failed\n"),
                              -1);
          TAO_AV_FlowSpecSetItor end = address_flow_set.end ();
          for (TAO_AV_FlowSpecSetItor start = address_flow_set.begin ();
               start != end; ++start)
            {
              TAO_FlowSpec_Entry *entry = (*start);
              switch (entry->direction ())
                {
                case TAO_FlowSpec_Entry::TAO_AV_DIR_IN:
                  {
                    if (entry->handler () != 0)
                      {
                        //Yamuna:PLEASE CHECK THIS LATER
#if defined ACE_HAS_RAPI || defined (ACE_HAS_WINSOCK2_GQOS)
                        // For IN flows on the A side we should remove the handlers from the reactor.
                        ACE_Event_Handler *event_handler = entry->handler ()->event_handler ();

			if (event_handler->reactor () != 0)
			  {
			    result = event_handler->reactor ()->remove_handler (event_handler,
										ACE_Event_Handler::READ_MASK);

			    if (result < 0)
			      if (TAO_debug_level > 0)
				ACE_DEBUG ((LM_DEBUG,
					    "TAO_AV_Core::init_forward_flows: remove_handler failed\n"));
			  }
#endif //ACE_HAS_RAPI || ACE_HAS_WINSOCK2_GQOS
                      }
                  }
                default:
                  break;
                }
              // Now if the address_set has been changed due to the addition of a control entry we should
              // add that to the flow_spec_set also.
              if (flow_spec_set.find (entry) < 0)
                {
                  // entry doesn't exist so add it.
                  flow_spec_set.insert (entry);
                  //                   size_t len = flow_spec.length ();
                  //                   flow_spec.length (len+1);
                  //                   flow_spec [len] = entry->entry_to_string ();
                }
            }
        }
      break;
    case TAO_AV_Core::TAO_AV_ENDPOINT_B:
      {
        if (address_flow_set.size () > 0)
          {
            ACE_DEBUG ((LM_DEBUG,
                        "(%N,%l) This connector registry is called\n"));

            result = this->connector_registry_->open (endpoint,
                                                      this,
                                                      address_flow_set);
            if (result == -1)
              ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_Core::init_Forward_flows: connector_registry open failed\n"),-1);
            TAO_AV_FlowSpecSetItor end = address_flow_set.end ();
            for (TAO_AV_FlowSpecSetItor start = address_flow_set.begin ();
                 start != end; ++start)
              {
                TAO_FlowSpec_Entry *entry = (*start);
                switch (entry->direction ())
                  {
                  case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                    {
		      if (entry->handler () != 0)
			{
			  // @@Naga: This wont be called in the case of Full Profile.
			  // For IN flows on the A side we should remove the handlers from the reactor.
			  ACE_Event_Handler *event_handler = entry->handler ()->event_handler ();
			  result = event_handler->reactor ()->remove_handler (event_handler,
									      ACE_Event_Handler::READ_MASK);
			  if (result < 0)
			    if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG,"TAO_AV_Core::init_forward_flows: remove_handler failed\n"));
			}
                    }
                  default:
                    break;
                  }
		// Now if the address_set has been changed due to the addition of a control entry we should
		// add that to the flow_spec_set also.
		if (flow_spec_set.find (entry) < 0)
		  {
		    // entry doesn't exist so add it.
		    flow_spec_set.insert (entry);
		  }
              }
          }
        if (flow_set.size () > 0)
          {
	    TAO_AV_FlowSpecSet tmp_flow_set (flow_set);
	    flow_set.reset ();
	    TAO_AV_FlowSpecSetItor end = tmp_flow_set.end ();
	    TAO_AV_FlowSpecSetItor start = tmp_flow_set.begin ();
            for (; start != end; ++start)
              {
		TAO_FlowSpec_Entry *entry = *start;
		TAO_FlowSpec_Entry *new_entry;
		ACE_CString dir;
		if (entry->direction () == 0)
		  dir += "IN";
		else if (entry->direction () == 1)
		  dir += "OUT";
		if (entry->get_peer_addr () != 0)
		  {
		    ACE_NEW_RETURN (new_entry,
				    TAO_Forward_FlowSpec_Entry (entry->flowname (),
								dir.c_str (),
								entry->format (),
								entry->flow_protocol_str (),
								entry->carrier_protocol_str (),
								entry->get_peer_addr (),
								entry->control_address ()),
				    -1);
		  }
		else
		  {
		    ACE_NEW_RETURN (new_entry,
				    TAO_Forward_FlowSpec_Entry (entry->flowname (),
								dir.c_str (),
								entry->format (),
								entry->flow_protocol_str (),
								entry->carrier_protocol_str (),
								entry->address (),
								entry->control_address ()),
				    -1);
		  }
		flow_set.insert (new_entry);
	      }
            result = this->acceptor_registry_->open (endpoint,
                                                     this,
                                                     flow_set);
            if (result == -1)
              ACE_ERROR_RETURN ((LM_ERROR,"TAO_AV_Core::init_Forward_flows: Acceptor_registry open failed\n"),-1);
            end = address_flow_set.end ();
	    start = address_flow_set.begin ();
            for (; start != end; ++start)
              {
                TAO_FlowSpec_Entry *entry = (*start);
                switch (entry->direction ())
                  {
                  case TAO_FlowSpec_Entry::TAO_AV_DIR_OUT:
                    {
		      if (entry->handler () != 0)
			{
			  // For IN flows on the A side we should remove the handlers from the reactor.
			  ACE_Event_Handler *event_handler = entry->handler ()->event_handler ();
			  result = event_handler->reactor ()->remove_handler (event_handler,
									      ACE_Event_Handler::READ_MASK);
			  if (result < 0)