application.cpp

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

// application.cpp,v 1.4 2003/11/25 10:54:31 jwillemsen Exp

#include "Constants.h"

#include "orbsvcs/orbsvcs/Event_Utilities.h"
#include "orbsvcs/orbsvcs/Event/EC_Lifetime_Utils_T.h"
#include "orbsvcs/orbsvcs/Event/ECG_UDP_Sender.h"
#include "orbsvcs/orbsvcs/Event/ECG_UDP_Receiver.h"

#include "orbsvcs/orbsvcs/RtecEventChannelAdminC.h"
#include "orbsvcs/orbsvcs/RtecEventCommS.h"

#include "tao/ORB_Core.h"

#include "ace/Array_Base.h"
#include "ace/Get_Opt.h"
#include "ace/Reactor.h"
#include "ace/os_include/os_netdb.h"

// Indicates whether this application is responsible for destroying
// the Event Channel it's using upon exit.
int destroy_ec_flag = 0;

/**
 * @class Heartbeat_Application
 *
 * @brief A simple application for testing federation of Event
 *        Channels via multicast.
 *
 *        NOTE: Contains platform-specific code (event data), i.e.,
 *              might not work cross-platform.
 *
 * This class acts both as a receiver and a supplier of HEARTBEAT events
 * to a multicast-federated Event Channel.  After sending a prespecified
 * number of heartbeat events, it prints out a summary about received
 * heartbeats and shuts down.
 */
class Heartbeat_Application :
  public virtual PortableServer::RefCountServantBase,
  public POA_RtecEventComm::PushConsumer,
  public TAO_EC_Deactivated_Object
{
public:

  /// Constructor.
  Heartbeat_Application (void);

  /// Destructor.
  ~Heartbeat_Application (void);

  // Initializes the object: connects with EC as a supplier and a
  // consumer and registers with reactor for timeouts.  If init ()
  // completes successfully, shutdown () must be called when this
  // object is no longer needed, for proper resource cleanup.  (This
  // is normally done by handle_timeout() method, but if handle_timeout()
  // will not have a chance to execute, it is the responsibility of
  // the user.)
  void init (CORBA::ORB_var orb,
             RtecEventChannelAdmin::EventChannel_var ec
             ACE_ENV_ARG_DECL);

  // No-op if the object hasn't been fully initialized.  Otherwise,
  // deregister from reactor and poa, destroy ec or just disconnect from it
  // (based on <destroy_ec> flag), and shut down the orb.
  void shutdown (void);

  /// Send another heartbeat or, if we already sent/attempted the required
  /// number of heartbeats, perform shutdown().
  int handle_timeout (const ACE_Time_Value& tv,
                      const void* act);

  /// PushConsumer methods.
  //@{
  /// Update our <heartbeats_> database to reflect newly received heartbeats.
  virtual void push (const RtecEventComm::EventSet &events
                     ACE_ENV_ARG_DECL)
    ACE_THROW_SPEC((CORBA::SystemException));

  /// Initiate shutdown().
  virtual void disconnect_push_consumer (ACE_ENV_SINGLE_ARG_DECL)
    ACE_THROW_SPEC((CORBA::SystemException));
  //@}

private:

  /**
   * @class Timeout_Handler
   *
   * @brief Helper class for receiving timeouts from Reactor.
   */
  class Timeout_Handler : public ACE_Event_Handler
  {
  public:
    /// Constructor.
    Timeout_Handler (Heartbeat_Application *recv);
    /// Reactor callback.
    virtual int handle_timeout (const ACE_Time_Value& tv,
                                const void* act);
  private:
    /// We callback to this object when a message arrives.
    Heartbeat_Application* receiver_;
  };

  /// Helpers.
  //@{
  /// Verify that arguments are not nil and store their values.
  int check_args (CORBA::ORB_var orb,
                  RtecEventChannelAdmin::EventChannel_var ec);
  /// Connects to EC as a supplier.
  void connect_as_supplier (ACE_ENV_SINGLE_ARG_DECL);
  /// Connects to EC as a consumer.  Activate with default POA.
  void connect_as_consumer (ACE_ENV_SINGLE_ARG_DECL);
  /// Call destroy() on the EC.  Does not propagate exceptions.
  void destroy_ec (void);
  /// Registers with orb's reactor for timeouts ocurring every 0.5
  /// seconds. Returns 0 on success, -1 on error.
  int register_for_timeouts (void);
  /// Deregister from reactor.
  void stop_timeouts (void);
  //@}

  /// Flag indicating whether this object has been fully initialized.
  int initialized_;

  /// Helper object for receiving timeouts from Reactor.
  Timeout_Handler timeout_handler_;

  /// Number of heartbeats we sent so far.
  size_t n_timeouts_;

  /// Info we keep on each HEARTBEAT source.
  typedef struct {
    pid_t pid;
    char hostname [MAXHOSTNAMELEN];
    int total;
  } HEARTBEAT_SOURCE_ENTRY;

  /// Stores info on all heartbeats we received so far.
  ACE_Array_Base<HEARTBEAT_SOURCE_ENTRY> heartbeats_;

  /// Our identity: pid followed by hostname.  We include this info into each
  /// heartbeat we send.
  char hostname_and_pid_ [MAXHOSTNAMELEN+11];

  /// ORB and EC pointers - to allow cleanup down the road.
  CORBA::ORB_var orb_;
  RtecEventChannelAdmin::EventChannel_var ec_;

  /// Consumer proxy which represents us in EC as a supplier.
  RtecEventChannelAdmin::ProxyPushConsumer_var consumer_;

  typedef TAO_EC_Auto_Command<TAO_ECG_UDP_Sender_Disconnect_Command>
  Supplier_Proxy_Disconnect;
  typedef TAO_EC_Auto_Command<TAO_ECG_UDP_Receiver_Disconnect_Command>
  Consumer_Proxy_Disconnect;

  /// Manages our connection to Supplier Proxy.
  Supplier_Proxy_Disconnect supplier_proxy_disconnect_;
  /// Manages our connection to Consumer Proxy.
  Consumer_Proxy_Disconnect consumer_proxy_disconnect_;
};
// **************************************************************************

Heartbeat_Application::Timeout_Handler::
Timeout_Handler (Heartbeat_Application* r)
  :  receiver_ (r)
{
}

int
Heartbeat_Application::Timeout_Handler::
handle_timeout (const ACE_Time_Value& tv,
                const void* act)
{
  return this->receiver_->handle_timeout (tv, act);
}

// **************************************************************************

Heartbeat_Application::Heartbeat_Application (void)
  : initialized_ (0)
  , timeout_handler_ (this)
  , n_timeouts_ (0)
  , orb_ ()
  , ec_ ()
  , consumer_ ()
  , supplier_proxy_disconnect_ ()
  , consumer_proxy_disconnect_ ()
{
}

Heartbeat_Application::~Heartbeat_Application (void)
{
}

int
Heartbeat_Application::check_args (CORBA::ORB_var orb,
                                   RtecEventChannelAdmin::EventChannel_var ec)
{
  if (CORBA::is_nil (ec.in ()))
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         "%N (%l): Nil ec argument to "
                         "Heartbeat_Application::init\n"),
                        -1);
    }

  if (CORBA::is_nil (orb.in ()))
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         "%N (%l): Nil orb argument to "
                         "Heartbeat_Application::init\n"),
                        -1);
    }

  this->ec_ = ec;
  this->orb_ = orb;

  return 0;
}

void
Heartbeat_Application::init (CORBA::ORB_var orb,
                             RtecEventChannelAdmin::EventChannel_var ec
                             ACE_ENV_ARG_DECL)
{
  // Verify arguments.
  if (this->check_args (orb, ec) == -1)
    {
      ACE_THROW (CORBA::INTERNAL ());
    }

  // Get hostname & process id, i.e., identity of this application.
  pid_t pid = ACE_OS::getpid ();
  ACE_OS::memcpy (this->hostname_and_pid_,
                  &pid,
                  sizeof (pid));

  if (gethostname (this->hostname_and_pid_ + sizeof (pid),
                   MAXHOSTNAMELEN)
      != 0)
    {
      ACE_ERROR ((LM_ERROR,
                  "Heartbeat_Application::init - "
                  "cannot get hostname\n"));
      ACE_THROW (CORBA::INTERNAL ());
    }

  // Connect to EC as a supplier.
  this->connect_as_supplier (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  // Connect to EC as a consumer.
  ACE_TRY
    {
      this->connect_as_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
      ACE_TRY_CHECK;
    }
  ACE_CATCHANY
    {
      this->consumer_proxy_disconnect_.execute ();
      ACE_RE_THROW;
    }
  ACE_ENDTRY;
  ACE_CHECK;

  // Register for reactor timeouts.
  if (this->register_for_timeouts () == -1)
    {
      this->consumer_proxy_disconnect_.execute ();
      this->supplier_proxy_disconnect_.execute ();
      this->deactivator_.deactivate ();
      ACE_THROW (CORBA::INTERNAL ());
    }

  this->initialized_ = 1;
}

int
Heartbeat_Application::register_for_timeouts (void)
{
  // Schedule timeout every 0.5 seconds, for sending heartbeat events.
  ACE_Time_Value timeout_interval (0, 500000);
  ACE_Reactor *reactor = this->orb_->orb_core ()->reactor ();
  if (!reactor
      || reactor->schedule_timer (&this->timeout_handler_, 0,
                                  timeout_interval,
                                  timeout_interval) == -1)
    {
      ACE_ERROR_RETURN ((LM_ERROR,
                         "Heartbeat_Application::register_for_timeouts - "
                         "cannot schedule timer\n"),
                        -1);
    }

  return 0;
}

void
Heartbeat_Application::stop_timeouts (void)
{
  ACE_Reactor *reactor = this->orb_->orb_core ()->reactor ();
  if (!reactor
      || reactor->cancel_timer (&this->timeout_handler_) == -1)
    {
      ACE_ERROR ((LM_ERROR,
                  "Heartbeat_Application::stop_timeouts - "
                  "cannot deregister from reactor.\n"));
    }
}

void
Heartbeat_Application::connect_as_supplier (ACE_ENV_SINGLE_ARG_DECL)
{
  // Obtain reference to SupplierAdmin.
  RtecEventChannelAdmin::SupplierAdmin_var supplier_admin =
    this->ec_->for_suppliers (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;

  // Obtain ProxyPushConsumer and connect this supplier.
  RtecEventChannelAdmin::ProxyPushConsumer_var proxy =
    supplier_admin->obtain_push_consumer (ACE_ENV_SINGLE_ARG_PARAMETER);
  ACE_CHECK;
  Consumer_Proxy_Disconnect new_proxy_disconnect (proxy.in ());

  ACE_SupplierQOS_Factory qos;
  qos.insert (SOURCE_ID, HEARTBEAT, 0, 1);

  proxy->connect_push_supplier (RtecEventComm::PushSupplier::_nil (),
                                qos.get_SupplierQOS ()
                                ACE_ENV_ARG_PARAMETER);
  ACE_CHECK;

  // Update resource managers.