event_channel.h

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

/* -*- C++ -*- */
//=============================================================================
/**
 *  @file Event_Channel.h
 *
 *  Event_Channel.h,v 1.45 2003/10/28 18:34:19 bala Exp
 *
 *  @author Tim Harrison ([e-mail])
 *
 *  TAO implementation of the Real Time Event Services. For more
 *  detailed information, see
 *  http://www.cs.wustl.edu/~schmidt/oopsla.ps.gz
 *
 * =  NAMING CONVENTIONS
 *  Some of the naming might be confusing.  For instance
 *  ACE_Push_Consumer_Proxy "is-a" ProxyPushSupplier.  To the
 *  channel, ACE_Push_Consumer_Proxy is a proxy to push consumers.
 *  To a push consumer, ACE_Push_Consumer_Proxy is a proxy to push
 *  suppliers.  I chose to name classes relative to the Event
 *  Channel.
 *
 *
 */
//=============================================================================


#ifndef ACE_EVENT_CHANNEL_H
#define ACE_EVENT_CHANNEL_H

#include /**/ "ace/pre.h"

#include "ace/Unbounded_Set_Ex.h"

#if !defined (ACE_LACKS_PRAGMA_ONCE)
# pragma once
#endif /* ACE_LACKS_PRAGMA_ONCE */

#include "ace/Map_Manager.h"
#include "ace/Functor.h"

#include "tao/Timeprobe.h"

#include "orbsvcs/Scheduler_Factory.h"
#include "orbsvcs/Event/Local_ESTypes.h"
#include "orbsvcs/Event/Timer_Module.h"
#include "orbsvcs/Event/ReactorTask.h"
#include "orbsvcs/Event/Event_Manip.h"

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

/**
 * An array of Events.  The Event Channel keeps several collections of
 * TAO_EC_Event objects, this is implemented using a simple Event Array.
 */
typedef ACE_Array<TAO_EC_Event> TAO_EC_Event_Array;

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

/**
 * Append an event to an event Array.  Simplify a common idiom when
 * manipulating arrays of events.
 */
ACE_INLINE
void operator += (TAO_EC_Event_Array &dest,
                  const TAO_EC_Event &item);

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

/**
 * Compare two events.  The Event Channel must compare events
 * (actually event headers) for equality.
 */
ACE_INLINE int operator == (const RtecEventComm::Event &event1,
                            const RtecEventComm::Event &event2);

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

/**
 * Compare two TAO_EC_Events.  This is only used in the instantiation of
 * ACE_Array<TAO_EC_Event>.
 */
ACE_INLINE
int operator != (const TAO_EC_Event &rhs, const TAO_EC_Event &lhs);

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

/**
 * Utility for debugging events.
 */
void TAO_RTOLDEvent_Export dump_event (const RtecEventComm::Event &event);

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


#if defined(_MSC_VER)
#if (_MSC_VER >= 1200)
#pragma warning(push)
#endif /* _MSC_VER >= 1200 */
#pragma warning(disable:4250)
#endif /* _MSC_VER */

/**
 * @class ACE_RTU_Manager
 *
 * @brief ACE RTU Manager
 *
 */
class TAO_RTOLDEvent_Export ACE_RTU_Manager
{
public:
  /// If <active> == 0, everything returns 0.  If <active> != 0, RTUs
  /// galore.
  ACE_RTU_Manager (int active);

  /// Returns 1 if the current task should preempt itself.  Otherwise,
  /// returns 0.  Resets should_preempt to zero.
  int should_preempt (void);

  /// Called by the dispatching module when the current task should
  /// preempt itself.
  void should_preempt (int s);

  /// If <nd> != 0, the current running task will be enqueued at the
  /// head of its dispatch tail.
  void not_done (int nd);

  /// Returns 1 if the current task needs to be dispatched again.
  /// Resets not_done_ to 0;
  int not_done (void);

  /// Get the priority of the current running task.
  RtecScheduler::OS_Priority priority (void);

  /// Set the priority of the current running task.
  void priority (RtecScheduler::OS_Priority priority);

private:
  int active_;
  int should_preempt_;
  int not_done_;
  RtecScheduler::OS_Priority priority_;
};

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

// Chesire cat.
// Forward declarations.
class ACE_ES_Priority_Timer;
class ACE_ES_Consumer_Module;
// This forward decl and typedef allow us to remove inheritence later
// on without changing any code.
//
class ACE_ES_Correlation_Module;
class ACE_ES_Subscription_Module;
class ACE_ES_Supplier_Module;
class ACE_ES_Dispatching_Base;
typedef ACE_ES_Dispatching_Base ACE_ES_Dispatching_Module;

// Forward declare the class used to connect several EC together.
class TAO_EC_Gateway;

// Factory class for the modules in the EC.
class TAO_Module_Factory;

/**
 * @class ACE_EventChannel
 *
 * @brief TAO's Real-time Event Channel.
 *
 * This class implements the interface defined in
 * RtecEventChannelAdmin.idl. For more details check:
 * http://www.cs.wustl.edu/~coryan/EC/JSAC98.pdf
 */
class TAO_RTOLDEvent_Export ACE_EventChannel : public POA_RtecEventChannelAdmin::EventChannel
{
public:
  enum { INITIAL_STATE = 0,
         CONSUMER = 1, SUPPLIER = 2,
         SHUTDOWN = CONSUMER | SUPPLIER };

  /**
   * Construction of the given <type>.  Check the **_CHANNEL
   * enumerations defined below.
   * By default we activate the threads on construction, but it is
   * possible to create the EC first and activate the threads later.
   * A factory for the modules can be provided, by default it uses
   * TAO_EC_Default_Module_Factory
   * If an scheduler is not provided it uses the singleton in
   * ACE_Scheduler_Factory.
   */
  ACE_EventChannel (CORBA::Boolean activate_threads = 1,
                    u_long type = ACE_DEFAULT_EVENT_CHANNEL_TYPE,
                    TAO_Module_Factory* factory = 0);
  ACE_EventChannel (RtecScheduler::Scheduler_ptr scheduler,
                    CORBA::Boolean activate_threads = 1,
                    u_long type = ACE_DEFAULT_EVENT_CHANNEL_TYPE,
                    TAO_Module_Factory* factory = 0);

  /// Calls destroy.
  virtual ~ACE_EventChannel (void);

  /// Allow transformations to RtecEventChannelAdmin::EventChannel.
  RtecEventChannelAdmin::EventChannel_ptr get_ref (ACE_ENV_SINGLE_ARG_DECL_NOT_USED);

  /// Returns a reference to the RTU manager.
  ACE_RTU_Manager *rtu_manager (void);

  // = These should be private.
  ACE_ES_Consumer_Module *consumer_module_;
  ACE_ES_Dispatching_Module *dispatching_module_;
  ACE_ES_Correlation_Module *correlation_module_;
  ACE_ES_Subscription_Module *subscription_module_;
  ACE_ES_Supplier_Module *supplier_module_;

  /// Consumer or supplier connected.
  void report_connect (u_long);

  /// Consumer or supplier disconnected.
  void report_disconnect (u_long);

  /// Activate the internal threads of the EC
  void activate (void);

  /// Do not call this.  The last module has shut down.
  void shutdown (void);

  /// Consumer or supplier connected.
  void report_connect_i (u_long);

  /// Consumer or supplier disconnected.
  void report_disconnect_i (u_long);

  /// Add gateways from the EC.
  void add_gateway (TAO_EC_Gateway* gw ACE_ENV_ARG_DECL);

  /// Remove gateways from the EC.
  void del_gateway (TAO_EC_Gateway* gw ACE_ENV_ARG_DECL);

  /// The consumer list has changed, thus the EC has to
  /// inform any gateways it has.
  void update_consumer_gwys (ACE_ENV_SINGLE_ARG_DECL);

  /// The supplier list has changed, thus the EC has to
  /// inform any gateways it has.
  void update_supplier_gwys (ACE_ENV_SINGLE_ARG_DECL);

  /// The timer module controls the strategy to dispatch timers.
  TAO_EC_Timer_Module* timer_module (void) const;

  // = The RtecEventChannelAdmin::EventChannel methods.

  /// In this implementation of the EC this returns the interface for
  /// the Consumer_Module.
  virtual RtecEventChannelAdmin::ConsumerAdmin_ptr
    for_consumers (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
      ACE_THROW_SPEC ((CORBA::SystemException));

  /// Return an interface to the Supplier_Module.
  virtual RtecEventChannelAdmin::SupplierAdmin_ptr
    for_suppliers (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
      ACE_THROW_SPEC ((CORBA::SystemException));

  /// Shutdown the EC, free all resources, stop all threads and then
  /// shutdown the server where the Servant is running.
  virtual void destroy (ACE_ENV_SINGLE_ARG_DECL_NOT_USED)
    ACE_THROW_SPEC ((CORBA::SystemException));

  /// The observer manipulators
  virtual RtecEventChannelAdmin::Observer_Handle
    append_observer (RtecEventChannelAdmin::Observer_ptr observer
                     ACE_ENV_ARG_DECL)
    ACE_THROW_SPEC ((
        CORBA::SystemException,
        RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR,
        RtecEventChannelAdmin::EventChannel::CANT_APPEND_OBSERVER));
  virtual void remove_observer (RtecEventChannelAdmin::Observer_Handle
                                ACE_ENV_ARG_DECL)
    ACE_THROW_SPEC ((
        CORBA::SystemException,
        RtecEventChannelAdmin::EventChannel::SYNCHRONIZATION_ERROR,
        RtecEventChannelAdmin::EventChannel::CANT_REMOVE_OBSERVER));

  // = Timer managment
  /// Schedule a timer at the appropriate priority for <preemption_priority>.
  /// Returns the preemption priority used on success, -1 on failure.
  int schedule_timer (RtecScheduler::handle_t rt_info,
                      const ACE_Command_Base *act,
                      RtecScheduler::OS_Priority preemption_priority,
                      const RtecScheduler::Time& delta,
                      const RtecScheduler::Time& interval = ORBSVCS_Time::zero ());

  /**
   * Cancel the timer associated with the priority of
   * <preemption_priority> and <id>.  <act> is filled in with the
   * Timer_ACT used when scheduling the timer.  Returns 0 on success,
   * -1 on failure.
   */
  int cancel_timer (RtecScheduler::OS_Priority preemption_priority,
                    int id,
                    ACE_Command_Base *&act);

  /**
   * Return a reference to its SchedulerService, notice that it uses
   * the CORBA semantics for memory managment, i.e. the user gains
   * ownership of the reference returned.
   */
  RtecScheduler::Scheduler_ptr scheduler (void);

  /**
   * @struct Observer_Entry
   *
   * @brief The data kept for each observer.
   *
   * The observer and its handle are kept in a simple structure.
   * In the future this structure could contain QoS information,
   * such as:
   * + how often do we update the observer?
   * + When was the last update.
   * + Does it want to receive all changes?
   */
  struct Observer_Entry
  {

    Observer_Entry (void);
    Observer_Entry (RtecEventChannelAdmin::Observer_Handle h,
                    RtecEventChannelAdmin::Observer_ptr o);

    /// The handle
    RtecEventChannelAdmin::Observer_Handle handle;

    /// The observer
    RtecEventChannelAdmin::Observer_var observer;

  };

private:
  /// Factor out commonality in the constructor.
  void init (int activate_threads);

  /// Remove all the observers, this simplifies the shutdown process.
  void cleanup_observers (void);

  /// The RTU manager dude!
  ACE_RTU_Manager *rtu_manager_;

  /// Can be any **_CHANNEL. (well, except NO_CHANNEL).
  u_long type_;

  /// Can be INITIAL_STATE, NO_CONSUMERS, NO_SUPPLIERS, or SHUTDOWN.
  u_long state_;

  /// Used to lock shared state.
  ACE_ES_MUTEX lock_;

  /// Ensures this->destory is executed only once.
  int destroyed_;

  typedef ACE_Map_Manager<RtecEventChannelAdmin::Observer_Handle,Observer_Entry,ACE_Null_Mutex> Observer_Map;
  typedef ACE_Map_Iterator<RtecEventChannelAdmin::Observer_Handle,Observer_Entry,ACE_Null_Mutex> Observer_Map_Iterator;

  /// The handles are generated in sequential order, but are opaque to
  /// the client.
  RtecEventChannelAdmin::Observer_Handle handle_generator_;

  /// Keep the set of Gateways, i.e. connections to peer EC.
  Observer_Map observers_;

  /// The strategy to dispatch timers.
  TAO_EC_Timer_Module* timer_module_;

  /// If 1 then we created the factory, thus we have to destroy it.
  int own_factory_;

  /// This is the factory we use to create and destroy the Event
  /// Channel modules.