event_sup.cpp

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

// Event_Sup.cpp,v 1.39 2003/11/04 08:12:59 dhinton Exp

// ============================================================================
//
// = FILENAME
//    Event_Sup.cpp
//
// = DESCRIPTION
//   Event Supplier for the flight simulator
//
// = AUTHOR
//    originally
//    David Levine ([e-mail]) and
//    Tim Harrison ([e-mail])
//    modified
//    Michael Kircher ([e-mail])
//
// ============================================================================

#include "Event_Sup.h"
#include "NavWeapC.h"

#include "orbsvcs/Event_Utilities.h"
#include "orbsvcs/Event_Service_Constants.h"
#include "orbsvcs/Scheduler_Factory.h"
#include "orbsvcs/RtecEventChannelAdminC.h"

#include "tao/PortableServer/ORB_Manager.h"

#include "ace/Get_Opt.h"
#include "ace/Sched_Params.h"
#include "ace/OS_NS_unistd.h"
#include "ace/OS_NS_stdio.h"
#include "ace/OS_NS_string.h"

#include "ace/os_include/os_ctype.h"

ACE_RCSID (Event_Supplier, 
           Event_Sup, 
           "Event_Sup.cpp,v 1.39 2003/11/04 08:12:59 dhinton Exp")

static const char usage [] =
"[[-?]\n"
"                 [-O[RBport] ORB port number]\n"
"                 [-m <count> of messages to send [100]\n"
"                 [-f name of schedler input data file]]\n";


Event_Supplier::Event_Supplier (int argc, char** argv)
: argc_(argc),
  argv_(argv),
  total_messages_(10),
  input_file_name_(0)
{
  navigation_.roll = navigation_.pitch = 0;
}

Event_Supplier::~Event_Supplier ()
{
  this->dOVE_Supplier_.disconnect ();
}

int
Event_Supplier::init ()
{
  this->get_options (argc_, argv_);
  return this->dOVE_Supplier_.connect ();
}

void
Event_Supplier::start_generating_events (void)
{
  unsigned long total_sent = 0;

  // Load the scheduling data for the simulation.
  ACE_Unbounded_Queue<Schedule_Viewer_Data *> schedule_data;
  this->load_schedule_data (schedule_data);
  ACE_Unbounded_Queue_Iterator<Schedule_Viewer_Data *>
    schedule_iter (schedule_data);

  if (schedule_iter.done ())
    {
      ACE_ERROR ((LM_ERROR,
                  "Event_Supplier::start_generating_events: "
                  "there is no scheduling data\n"));
      return;
    }

  CORBA::Any any;

  do
  {

  // Insert the event data
    this->insert_event_data (any,
                             schedule_iter);

    // deliver it over the wire
    dOVE_Supplier_.notify (any);

    if (total_sent < 5)
      ACE_DEBUG ((LM_DEBUG,
                  "Pushing event data.\n"));
    else if (total_sent == 5)
      ACE_DEBUG ((LM_DEBUG,
                  "Everything is running. Going to be mute.\n"));
  }
  while (++total_sent < this->total_messages_);

  // clean up the scheduling data
  Schedule_Viewer_Data **data_temp;
  for (schedule_iter.first ();
       schedule_iter.done () == 0;
       schedule_iter.advance ())
    if (schedule_iter.next (data_temp) && data_temp)
      delete (*data_temp);
}

void
Event_Supplier::load_schedule_data
      (ACE_Unbounded_Queue<Schedule_Viewer_Data *> &schedule_data)
{
  Schedule_Viewer_Data *data = 0;

  if (this->input_file_name_)
    {
      // Open the scheduler data input file and read its contents into
      // a queue.
      FILE *input_file;

      int scan_count = 0;
      input_file = ACE_OS::fopen(this->input_file_name_, "r");

      if (input_file)
        {
          // Get a line at a time from the data file and parse it.
          char input_buf[BUFSIZ];
          while (ACE_OS::fgets (input_buf, BUFSIZ, input_file))
            {
              // Run through leading whitespace.
              char *temp = input_buf;
              while (*temp && isspace (*temp))
                ++temp;

              // If there is anything besides whitespace in the line
              // read, scan its fields into the scheduling data
              // structure.
              if (ACE_OS::strlen (temp) > 0)
                {
                  ACE_NEW (data, Schedule_Viewer_Data);
                  scan_count = sscanf (temp, "%s %lf %lf %lu %lu %lu %lu",
                                       data->operation_name,
                                       &data->utilitzation,
                                       &data->overhead,
                                       &data->arrival_time,
                                       &data->deadline_time,
                                       &data->completion_time,
                                       &data->computation_time);
                  if (scan_count != 7)
                    {
                      ACE_ERROR ((LM_ERROR,
                                  "Event_Supplier::start_generating_events: "
                                  "scanned incorrect number of data elements: %d\n", scan_count));

                      delete data;
                      return;
                    }

                  // Insert the data into the queue.
                  schedule_data.enqueue_tail (data);
                }
            }
        }
      else
        {
          ACE_ERROR ((LM_ERROR,
                      "Event_Supplier::start_generating_events: "
                      "could not open input file [%s].\n",
                      this->input_file_name_));
          return;
        }
    }
  else
  {
    u_long last_completion = 0;

    // Just create 10 dummy scheduling records and use them.
    for (int i = 0; i < 10; ++i)
    {
      ACE_NEW (data, Schedule_Viewer_Data);

      const char *oper_name = 0;
      switch (i % 4)
      {
      case 0:
        oper_name = "high_20";
        break;

      case 1:
        oper_name = "low_20";
        break;

      case 2:
        oper_name = "high_10";
        break;

      case 3:
      default:
        oper_name = "low_10";
        break;
      }

      ACE_OS::strncpy (data->operation_name,
                       oper_name,
                       BUFSIZ-1);


      data->utilitzation = (double)(20.0+ACE_OS::rand() %10);
      data->overhead = (double)(ACE_OS::rand() %20);

      data->arrival_time = ACE_OS::rand() % 200;
      data->computation_time = (ACE_OS::rand() % 100) + 10;

      data->completion_time = last_completion + (ACE_OS::rand() % 100) + 100;
      data->completion_time =
        data->completion_time <  data->arrival_time + data->computation_time
        ? data->arrival_time + data->computation_time
        : data->completion_time;

      last_completion = data->completion_time;

      data->deadline_time = data->completion_time + (ACE_OS::rand() % 200) - 50;

      // insert the data into the queue.
      schedule_data.enqueue_tail (data);
    }
  }
}

// This function fills in the random data into the anys transported by
// the event channel.

void
Event_Supplier::insert_event_data (CORBA::Any &data,
                                  ACE_Unbounded_Queue_Iterator<Schedule_Viewer_Data *> &schedule_iter)
{
  static u_long last_completion = 0;