sender.cpp

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

// sender.cpp,v 1.11 2003/08/26 19:32:14 yamuna Exp

#include "sender.h"
#include "tao/debug.h"
#include "ace/Get_Opt.h"
#include "ace/High_Res_Timer.h"

static u_long peak_bandwidth = 18400;

typedef ACE_Singleton<Sender, ACE_Null_Mutex> SENDER;
// Create a singleton instance of the Sender.

// The time that should lapse between two consecutive frames sent.
ACE_Time_Value inter_frame_time;

CORBA::Boolean
Sender_StreamEndPoint::modify_QoS (AVStreams::streamQoS &new_qos,
                                   const AVStreams::flowSpec &the_flows
                                   ACE_ENV_ARG_DECL)
                                   ACE_THROW_SPEC (( CORBA::SystemException,
                                                     AVStreams::noSuchFlow,
                                                     AVStreams::QoSRequestFailed ))
{
  ACE_DEBUG ((LM_DEBUG,
              "Sender_StreamEndPoint::modify_QoS\n"));

  int result =  this->change_qos (new_qos, the_flows ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (0);

  if (result != 0)
    return 0;

  return 1;
}

int
Sender_StreamEndPoint::get_callback (const char *,
                                     TAO_AV_Callback *&callback)
{
  ACE_DECLARE_NEW_CORBA_ENV;
  // Create and return the sender application callback to AVStreams
  // for further upcalls.
  callback = &this->callback_;

  TAO_Negotiator *negotiator;
  ACE_NEW_RETURN (negotiator,
                  TAO_Negotiator,
                  -1);
  AVStreams::Negotiator_var negotiator_obj =
    negotiator->_this (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  this->set_negotiator (negotiator_obj.in ());

  return 0;
}

int
Sender_StreamEndPoint::set_protocol_object (const char *,
                                            TAO_AV_Protocol_Object *object)
{
  // Set the sender protocol object corresponding to the transport
  // protocol selected.
  SENDER::instance ()->protocol_object (object);
  return 0;
}

Sender::Sender (void)
  : sender_mmdevice_ (0),
    streamctrl_ (0),
    frame_count_ (0),
    filename_ ("input"),
    input_file_ (0),
    protocol_ ("QoS_UDP"),
    frame_rate_ (1.0),
    mb_ (BUFSIZ),
    address_ (0),
    peer_addr_ (0)
{
}

void
Sender::protocol_object (TAO_AV_Protocol_Object *object)
{
  // Set the sender protocol object corresponding to the transport
  // protocol selected.
  this->protocol_object_ = object;
}

int
Sender::parse_args (int argc,
                    char **argv)
{
  // Parse command line arguments
  ACE_Get_Opt opts (argc, argv, "f:p:r:l:a:d");

  int c;
  while ((c= opts ()) != -1)
    {
      switch (c)
        {
        case 'f':
          this->filename_ = opts.opt_arg ();
          break;
        case 'p':
          this->protocol_ = opts.opt_arg ();
          break;
        case 'r':
          this->frame_rate_ = (double)ACE_OS::atoi (opts.opt_arg ());
          break;
        case 'd':
          TAO_debug_level++;
          break;
        case 'l':
          this->address_ = opts.opt_arg ();
          break;
        case 'a':
          this->peer_addr_ = opts.opt_arg ();
          break;
        default:
          ACE_DEBUG ((LM_DEBUG, "Unknown Option\n"));
          return -1;
        }
    }
  return 0;
}

void
Sender::fill_qos (AVStreams::streamQoS &qos)
{
  peak_bandwidth += 100;

  ACE_DEBUG ((LM_DEBUG,
              "Sender::fill_qos %d\n",
              peak_bandwidth));

  qos.length (1);
  qos [0].QoSType =  CORBA::string_dup ("Data_Receiver");

  qos [0].QoSParams.length (10);

  qos [0].QoSParams [0].property_name = CORBA::string_dup ("Service_Type");
  qos [0].QoSParams [0].property_value <<= (CORBA::Short) 1;//ACE_SERVICETYPE_CONTROLLEDLOAD;

  qos [0].QoSParams [1].property_name = CORBA::string_dup ("Token_Rate");
  qos [0].QoSParams [1].property_value <<= (CORBA::ULong) 9200 ;

  qos [0].QoSParams [2].property_name = CORBA::string_dup ("Token_Bucket_Size");
  qos [0].QoSParams [2].property_value <<= (CORBA::ULong) 708;

  qos [0].QoSParams [3].property_name = CORBA::string_dup ("Peak_Bandwidth");
  qos [0].QoSParams [3].property_value <<= (CORBA::ULong) peak_bandwidth;


  qos [0].QoSParams [4].property_name = CORBA::string_dup ("Latency");
  qos [0].QoSParams [4].property_value <<= (CORBA::ULong) 0;

  qos [0].QoSParams [5].property_name = CORBA::string_dup ("Delay_Variation");
  qos [0].QoSParams [5].property_value <<= (CORBA::ULong) 0;

  qos [0].QoSParams [6].property_name = CORBA::string_dup ("Max_SDU_Size");
  qos [0].QoSParams [6].property_value <<= (CORBA::ULong) 368;

  qos [0].QoSParams [7].property_name = CORBA::string_dup ("Minimum_Policed_Size");
  qos [0].QoSParams [7].property_value <<= (CORBA::ULong) 368;

  qos [0].QoSParams [8].property_name = CORBA::string_dup ("TTL");
  qos [0].QoSParams [8].property_value <<= (CORBA::ULong) 25;

  qos [0].QoSParams [9].property_name = CORBA::string_dup ("Priority");
  qos [0].QoSParams [9].property_value <<= (CORBA::ULong) 1;

}

// Method to get the object reference of the receiver
int
Sender::bind_to_receiver (ACE_ENV_SINGLE_ARG_DECL)
{
  CosNaming::Name name (1);
  name.length (1);
  name [0].id =
    CORBA::string_dup ("Receiver");

  // Resolve the receiver object reference from the Naming Service
  CORBA::Object_var receiver_mmdevice_obj =
    this->naming_client_->resolve (name
                                   ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  this->receiver_mmdevice_ =
    AVStreams::MMDevice::_narrow (receiver_mmdevice_obj.in ()
                                  ACE_ENV_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  if (CORBA::is_nil (this->receiver_mmdevice_.in ()))
    ACE_ERROR_RETURN ((LM_ERROR,
                       "Could not resolve Receiver_MMdevice in Naming service <%s>\n"),
                      -1);

  return 0;
}

int
Sender::init (int argc,
              char **argv
              ACE_ENV_ARG_DECL)
{
  // Initialize the endpoint strategy with the orb and poa.
  int result =
    this->endpoint_strategy_.init (TAO_AV_CORE::instance ()->orb (),
                                   TAO_AV_CORE::instance ()->poa ());
  if (result != 0)
    return result;

  // Initialize the naming services
  result =
    this->naming_client_.init (TAO_AV_CORE::instance ()->orb ());
  if (result != 0)
    return result;

  // Parse the command line arguments
  result =
    this->parse_args (argc,
                      argv);
  if (result != 0)
    return result;

  // Open file to read.
  this->input_file_ =
    ACE_OS::fopen (this->filename_.c_str (),
                   "r");

  if (this->input_file_ == 0)
    ACE_ERROR_RETURN ((LM_DEBUG,
                       "Cannot open input file %s\n",
                       this->filename_.c_str ()),
                      -1);
  else
    ACE_DEBUG ((LM_DEBUG,
                "File opened successfully\n"));

  // Resolve the object reference of the receiver from the Naming Service.
  result = this->bind_to_receiver (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK_RETURN (-1);

  if (result != 0)
    ACE_ERROR_RETURN ((LM_ERROR,
                       "(%P|%t) Error binding to the naming service\n"),
                      -1);

  this->flowname_ = "Data_Receiver";

  // Set the address of the ftp client.
  ACE_INET_Addr* addr;
  if (this->address_ != 0)
    ACE_NEW_RETURN (addr,
		    ACE_INET_Addr (this->address_),
		    -1);
  else
    {
      char buf [BUFSIZ];
      ACE_OS::hostname (buf,
			BUFSIZ);
      ACE_NEW_RETURN (addr,
		      ACE_INET_Addr ("5000",
				     buf),
		      -1);
    }

  // Create the forward flow specification to describe the flow.
  TAO_Forward_FlowSpec_Entry entry (this->flowname_.c_str (),
                                    "IN",
                                    "USER_DEFINED",
                                    "",
                                    this->protocol_.c_str (),
                                    addr);