avstreams_i.cpp

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

              this->vdev_a_->modify_QoS (new_qos, in_flowspec ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK;
            }

          if (out_flowspec.length () != 0)
            {
              this->vdev_b_->modify_QoS (new_qos, out_flowspec ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK;
            }
        }
      ACE_CATCHANY
        {
          ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_StreamCtrl::modify_QoS");
          return 0;
        }
      ACE_ENDTRY;
      ACE_CHECK_RETURN (0);

    }
  return 1;
}

// ----------------------------------------------------------------------
// TAO_MCastConfigIf
// ----------------------------------------------------------------------

TAO_MCastConfigIf::TAO_MCastConfigIf (void)
  :peer_list_iterator_ (peer_list_)
{
}

TAO_MCastConfigIf::~TAO_MCastConfigIf (void)
{
  //no-op
}

// In future this should be a multicast message instead of point-to-point unicasts.
CORBA::Boolean
TAO_MCastConfigIf::set_peer (CORBA::Object_ptr peer,
                             AVStreams::streamQoS & qos,
                             const AVStreams::flowSpec & flow_spec
                             ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   AVStreams::QoSRequestFailed,
                   AVStreams::streamOpFailed))
{
  ACE_TRY
    {
      Peer_Info *info;
      ACE_NEW_RETURN (info,
                      Peer_Info,
                      0);
      info->peer_ = AVStreams::VDev::_narrow (peer ACE_ENV_ARG_PARAMETER);
      ACE_TRY_CHECK;
      info->qos_ = qos;
      info->flow_spec_ = flow_spec;
      this->peer_list_.insert_tail (info);
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_MCastConfigIf::set_peer");
      return 0;
    }
  ACE_ENDTRY;
  ACE_CHECK_RETURN (0);
  return 1;
}

// In future this should be a multicast message instead of point-to-point unicasts.
void
TAO_MCastConfigIf::configure (const CosPropertyService::Property & a_configuration
                              ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  Peer_Info *info;
  ACE_TRY
    {
      for (this->peer_list_iterator_.first ();
           (info = this->peer_list_iterator_.next ()) != 0;
           this->peer_list_iterator_.advance ())
        {
          info->peer_->configure (a_configuration ACE_ENV_ARG_PARAMETER);
          ACE_TRY_CHECK;
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_MCastConfigIf::set_configure");
      return;
    }
  ACE_ENDTRY;
  ACE_CHECK;
}


void
TAO_MCastConfigIf::set_initial_configuration (const CosPropertyService::Properties &initial
                                              ACE_ENV_ARG_DECL_NOT_USED)
  ACE_THROW_SPEC ((CORBA::SystemException))
{
  this->initial_configuration_ = initial;
}

// In future this should be a multicast message instead of point-to-point unicasts.
void
TAO_MCastConfigIf::set_format (const char * flowName,
                               const char * format_name
                               ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   AVStreams::notSupported))
{
  Peer_Info *info;
  ACE_TRY
    {
      for (this->peer_list_iterator_.first ();
           (info = this->peer_list_iterator_.next ()) != 0;
           this->peer_list_iterator_.advance ())
        {
          if (this->in_flowSpec (info->flow_spec_, flowName))
            {
              info->peer_->set_format (flowName, format_name ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK;
            }
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_MCastConfigIf::set_format");
      return;
    }
  ACE_ENDTRY;
  ACE_CHECK;
}

// In future this should be a multicast message instead of point-to-point unicasts.
void
TAO_MCastConfigIf::set_dev_params (const char * flowName,
                                   const CosPropertyService::Properties & new_params
                                   ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   AVStreams::PropertyException,
                   AVStreams::streamOpFailed))
{
  Peer_Info *info;
  ACE_TRY
    {

      for (this->peer_list_iterator_.first ();
           (info = this->peer_list_iterator_.next ()) != 0;
           this->peer_list_iterator_.advance ())
        {
          if (this->in_flowSpec (info->flow_spec_, flowName))
            {
              info->peer_->set_dev_params (flowName, new_params ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK;
            }
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_MCastConfigIf::set_dev_params");
      return;
    }
  ACE_ENDTRY;
  ACE_CHECK;
}

int
TAO_MCastConfigIf::in_flowSpec (const AVStreams::flowSpec& flow_spec, const char *flow_name)
{
  size_t len = ACE_OS::strlen (flow_name);
  for (CORBA::ULong i = 0; i < flow_spec.length (); i++)
    if (ACE_OS::strncmp (flow_spec[i], flow_name, len) == 0)
      {
        return 1;
      }
  return 0;
}

// ----------------------------------------------------------------------
// TAO_Base_StreamEndPoint
// ----------------------------------------------------------------------

TAO_Base_StreamEndPoint::TAO_Base_StreamEndPoint (void)
{
}

TAO_Base_StreamEndPoint::~TAO_Base_StreamEndPoint (void)
{
}

int
TAO_Base_StreamEndPoint::handle_close (void)
{
  // This method should not be defined, but EGCS complains endlessly
  // about it.
  return -1;
}

int
TAO_Base_StreamEndPoint::handle_open (void)
{
  return 0;
}

int
TAO_Base_StreamEndPoint::handle_stop (const AVStreams::flowSpec &
                                      ACE_ENV_ARG_DECL_NOT_USED)
{
  return 0;
}

int
TAO_Base_StreamEndPoint::handle_start (const AVStreams::flowSpec &
                                       ACE_ENV_ARG_DECL_NOT_USED)
{
  return 0;
}

int
TAO_Base_StreamEndPoint::handle_destroy (const AVStreams::flowSpec &
                                         ACE_ENV_ARG_DECL_NOT_USED)
{
  return 0;
}

// The following function is for backward compatibility.
CORBA::Boolean
TAO_Base_StreamEndPoint::handle_preconnect (AVStreams::flowSpec &)
{
  return 1;
}

// The following function is for backward compatibility.
CORBA::Boolean
TAO_Base_StreamEndPoint::handle_postconnect (AVStreams::flowSpec &)
{
  return 1;
}

// The following function is for backward compatibility.
CORBA::Boolean
TAO_Base_StreamEndPoint::handle_connection_requested (AVStreams::flowSpec &
                                                 ACE_ENV_ARG_DECL_NOT_USED)
{
  return 1;
}

int
TAO_Base_StreamEndPoint::set_protocol_object (const char * /*flowname*/,
                                              TAO_AV_Protocol_Object * /*sfp_object*/)
{
  return -1;
}

int
TAO_Base_StreamEndPoint::get_callback (const char * /*flowname*/,
                                       TAO_AV_Callback *&/*sfp_callback*/)
{
  return -1;
}

int
TAO_Base_StreamEndPoint::get_control_callback (const char * /*flowname*/,
                                               TAO_AV_Callback *&/*sfp_callback*/)
{
  return -1;
}

void
TAO_Base_StreamEndPoint::set_flow_handler (const char *flowname,
                                           TAO_AV_Flow_Handler *handler)
{
  if(TAO_debug_level > 1)
  {
     ACE_DEBUG ((LM_DEBUG, "(%N,%l) TAO_Base_StreamEndPoint::set_flow_handler(), flowname: %s\n", flowname));
  }
  ACE_CString flow_name_key (flowname);
  if (this->flow_handler_map_.bind (flow_name_key, handler) != 0)
    ACE_ERROR ((LM_ERROR,
                "Error in storing flow handler\n"));
}

void
TAO_Base_StreamEndPoint::set_control_flow_handler (const char *flowname,
                                                   TAO_AV_Flow_Handler *handler)
{
  ACE_CString flow_name_key (flowname);
  if (this->control_flow_handler_map_.bind (flow_name_key, handler) != 0)
    ACE_ERROR ((LM_ERROR,
                "Error in storing control flow handler\n"));
}

// ----------------------------------------------------------------------
// TAO_StreamEndPoint
// ----------------------------------------------------------------------

// constructor.

TAO_StreamEndPoint::TAO_StreamEndPoint (void)
  :flow_count_ (0),
   flow_num_ (0),
   mcast_port_ (ACE_DEFAULT_MULTICAST_PORT+1)
{
  //is->mcast_addr_ = ACE_OS::inet_addr (ACE_DEFAULT_MULTICAST_ADDR);
  this->mcast_addr_.set (ACE_DEFAULT_MULTICAST_ADDR);
  if (TAO_debug_level > 0) ACE_DEBUG ((LM_DEBUG, "TAO_StreamEndPoint::TAO_StreamEndPoint::mcast_addr = %s", this->mcast_addr_.c_str ()));
  //  this->handle_open ();
}


CORBA::Boolean
TAO_StreamEndPoint::connect (AVStreams::StreamEndPoint_ptr responder,
                             AVStreams::streamQoS &qos,
                             const AVStreams::flowSpec &the_spec
                             ACE_ENV_ARG_DECL)
  ACE_THROW_SPEC ((CORBA::SystemException,
                   AVStreams::noSuchFlow,
                   AVStreams::QoSRequestFailed,
                   AVStreams::streamOpFailed))
{
  if (TAO_debug_level > 0)
    ACE_DEBUG ((LM_DEBUG, "TAO_StreamEndPoint::Connect ()\n"));
  CORBA::Boolean retv = 0;
  this->peer_sep_ = AVStreams::StreamEndPoint::_duplicate (responder);
  ACE_TRY_EX (negotiate)
    {
      if (!CORBA::is_nil (this->negotiator_.in ()))
        {
          ACE_DEBUG ((LM_DEBUG,
                      "NEGOTIATOR AVIALABLE\n"));

          CORBA::Any_var negotiator_any = responder->get_property_value ("Negotiator");

          AVStreams::Negotiator_ptr peer_negotiator;
          negotiator_any.in () >>= peer_negotiator;
          if (!CORBA::is_nil (peer_negotiator))
            {
              CORBA::Boolean result =
                this->negotiator_->negotiate (peer_negotiator,
                                              qos
                                              ACE_ENV_ARG_PARAMETER);
              ACE_TRY_CHECK_EX (negotiate);
              if (!result)
                if (TAO_debug_level > 0)
                  ACE_DEBUG ((LM_DEBUG, "TAO_StreamEndPoint::Connect (): negotiate failed\n"));
            }
        }
    }
  ACE_CATCHANY
    {
      ACE_PRINT_EXCEPTION (ACE_ANY_EXCEPTION, "TAO_StreamEndPoint::negotiate");
    }
  ACE_ENDTRY;
  ACE_CHECK_RETURN (0);

  ACE_TRY_EX (available_protocols)
    {
      if (this->protocols_.length () > 0)
        {
          // choose proto