event_channel.h

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

  TAO_Module_Factory* module_factory_;

  /// The scheduler
  RtecScheduler::Scheduler_var scheduler_;
};

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

/**
 * @class ACE_ES_Dependency_Iterator
 *
 * @brief ConsumerQOS Iterator
 *
 * This is used by the Event Channel to parse ConsumerDependency objects.
 */
class TAO_RTOLDEvent_Export ACE_ES_Dependency_Iterator
{
public:
  /// Construct and iterator for <rep>.
  ACE_ES_Dependency_Iterator (RtecEventChannelAdmin::DependencySet &rep);

  /// Returns 0 if the advance succeeded.  Returns -1 if there are no
  /// more dependencies in the group.
  int advance_dependency (void);

  /// Cache values for n_** methods.
  int parse (void);

  /// Returns the number of conjunction groups in the dependency set.
  int n_conjunctions (void);

  /// Returns the number of disjunction groups in the dependency set.
  int n_disjunctions (void);

  /// Returns the number of timeouts registered.
  int n_timeouts (void);

  /// Returns the number of events registered.
  int n_events (void);

  /// Accessor to the current ConsumerDependency pointed to by the
  /// iterator.
  RtecEventChannelAdmin::Dependency &operator *(void);

  /// Returns the first RT_Info in the dependencies.
  RtecScheduler::handle_t first_rt_info (void);

protected:
  /// The first rt_info in the dependencies.
  RtecScheduler::handle_t rt_info_;

  /// Reference to the dependency array.
  RtecEventChannelAdmin::DependencySet &rep_;

  /// Index into rep_.
  int index_;

  /// The type of the current correlation group.
  RtecEventComm::EventType group_type_;

  /// Number of conjunction groups.
  int n_conjunctions_;

  /// Number of disjunction groups.
  int n_disjunctions_;

  /// Number of timeouts registered.
  int n_timeouts_;

  /// Number of events registered.
  int n_events_;
};

// ************************************************************
// Forward decl.
class ACE_ES_Consumer_Rep_Timeout;

/**
 * @class ACE_ES_Disjunction_Group
 *
 * @brief Disjunction Group
 *
 * Represents a disjunction group, such as (A|B|C).
 */
class TAO_RTOLDEvent_Export ACE_ES_Disjunction_Group
{
public:
  /// Default construction.
  ACE_ES_Disjunction_Group (void);

  /// Destructor.
  virtual ~ACE_ES_Disjunction_Group (void);

  /// <cm> is needed for rescheduling deadlines.
  void set_correlation_module (ACE_ES_Correlation_Module *cm);

  /// If deadline_timer_rep_ is set, it is cancelled and rescheduled.
  void reschedule_deadline (void);

  /// Set the group's reference to the deadline timer.  Returns 0 on
  /// success, -1 on failure.
  int set_deadline_timeout (ACE_ES_Consumer_Rep_Timeout *cr);

  /// Does nothing.  This is the only virtual method in this little
  /// heirarchy with the conjunction group.
  virtual void add_events (TAO_EC_Event_Array *outbox,
                           TAO_EC_Event_Array *pending_events,
                           u_long &pending_flags);

  /// Set the ACT for this group.
  void set_act (RtecEventComm::Event &act);

protected:
  /// To be sent with this group.
  TAO_EC_Event act_;

private:
  /// The disjunction group keeps a reference to the deadline timer.
  ACE_ES_Consumer_Rep_Timeout *deadline_timer_rep_;

  /// Used for cancelling and scheduling deadline_timer_rep_.
  ACE_ES_Correlation_Module *correlation_module_;
};

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

/**
 * @class ACE_ES_Conjunction_Group
 *
 * @brief Conjunction Group
 *
 * Represents a conjunction group, such as (A+B+C).
 */
class TAO_RTOLDEvent_Export ACE_ES_Conjunction_Group : public ACE_ES_Disjunction_Group
{
public:
  /// Default construction.
  ACE_ES_Conjunction_Group (void);

  /// Destructor.
  virtual ~ACE_ES_Conjunction_Group (void);

  /// Set the <type_id>th bit in the forward_value_.
  int add_type (int type_id);

  /// Returns 1 if this conjunction group's dependencies have been
  /// satisfied.  Returns 0 otherwise.
  int should_forward (u_long pending_flags);

  /**
   * For each bit set in forward_value_, the corresponding events in
   * <pending_events> is added to <outbox>.  Each bit set in
   * <forward_value_> is cleared in <pending_flags>.  If <oubox> == 0,
   * then add_events just clears the pending events and flags.
   */
  virtual void add_events (TAO_EC_Event_Array *outbox,
                           TAO_EC_Event_Array *pending_events,
                           u_long &pending_flags);

private:
  u_long forward_value_;
};

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

// Forward decl.
class ACE_ES_Consumer_Correlation;

/**
 * @class ACE_ES_Consumer_Rep
 *
 * @brief Consumer Representation.
 *
 * These are stored in the subscription module.  They store
 * information that allows optimized correlations.  It represents
 * the consumer that will handle *one* type of event.  This
 * probably shouldn't inherit from ACE_Command_Base since it's used
 * only by ACE_ES_Consumer_Rep_Timeout.  However, this allows me to
 * minimize dynamic allocation.
 */
class TAO_RTOLDEvent_Export ACE_ES_Consumer_Rep : public ACE_Command_Base
{
public:
  /// Default construction.
  ACE_ES_Consumer_Rep (void);

  /// <dep> describes the event subscribed to and the method handling
  /// the event.   <correlation> is the parent correlation object.
  void init (ACE_ES_Consumer_Correlation *correlation,
             RtecEventChannelAdmin::Dependency &dep);

  /// Virtual destruction.
  virtual ~ACE_ES_Consumer_Rep (void);

  /// The event subscribed to and the method that will handle this
  /// event.
  RtecEventChannelAdmin::Dependency *dependency (void);

  /// Get the correlation group index of this consumer rep's event
  /// type.
  int type_id (void);

  /// Set the correlation group index of this consumer rep's event
  /// type.
  void type_id (int);

  enum Correlation_Type
  {
    NO_CORRELATION,
    CORRELATE,
    DEADLINE_TIMEOUT,
    GLOBAL_DEADLINE
  };

  /// If this returns 0, then the event associated with this consumer
  /// should be forwarded without running any correlations.
  u_long correlation_type (void);

  /// Set whether the event should be correlated.  <ct> is a
  /// Correlation_Type.
  void correlation_type (u_long ct);

  /// Add a disjunction group.
  int add_disjunction_group (ACE_ES_Disjunction_Group &);

  /// Returns the first disjunction group added via
  /// this->add_disjunction_group.
  ACE_ES_Disjunction_Group *top_group (void);

  /// Calls reschedule_deadline on all disjunction groups added through
  /// this->add_disjunction_group.
  void reschedule_deadlines (void);

  /// Returns 1 if events should be sent to this consumer.  Returns 0
  /// if they should not (suspended or disconnected).
  int receiving_events (void);

  /// Stop forwarding events to the calling consumer.
  void suspend (void);

  /// Resume forwarding events to the calling consumer.
  void resume (void);

  /// Returns the Consumer_Correlation object for the target consumer.
  ACE_ES_Consumer_Correlation *correlation (void);

  /// Schedules the consumer rep to be removed from all subscription
  /// lists.
  void disconnect (void);

  /// Returns 1 if the consumer rep should be removed from all
  /// subscription lists.
  int disconnected (void);

  /// Increments ref_count_.
  void _duplicate (void);

  /// Decrements ref_count_ and deletes this if 0.
  void _release (void);

protected:
  /// Whether the rep should be removed from all subscription lists.
  int disconnected_;

  /**
   * This is called when timeouts occur.  This implementation prints
   * out an error message (since it really shouldn't be implemented in
   * this class).
   */
  virtual int execute (void* arg = 0);

  /// Whether events should be dropped or forwarded.
  int suspended_;

  /// Whether any correlating should be done for this event.
  u_long correlation_type_;

  /// Event subscribed to.
  RtecEventChannelAdmin::Dependency *dependency_;

  /// The target consumer of events.
  ACE_ES_Consumer_Correlation *correlation_;

  /// Correlation group index of event_->type_.
  int type_id_;

  /// This should be a set.  We'll just have room for one now.
  ACE_ES_Disjunction_Group *disjunction_group_;

  /// Lock for reference count.
  ACE_Atomic_Op<ACE_ES_MUTEX, int> ref_count_;
};

/**
 * @class ACE_ES_Consumer_Rep_Timeout
 *
 * @brief Consumer Representation.
 *
 * These are stored in the subscription module.  They store
 * information that allows optimized correlations.  It represents
 * the consumer that will handle *one* type of event.
 */
class TAO_RTOLDEvent_Export ACE_ES_Consumer_Rep_Timeout : public ACE_ES_Consumer_Rep
{
public:
  /// Default construction.
  ACE_ES_Consumer_Rep_Timeout (void);

  /// <dep> describes the event subscribed to and the method handling
  /// the event.   <correlation> is the parent correlation object.
  void init (ACE_ES_Consumer_Correlation *correlation,
             RtecEventChannelAdmin::Dependency &dep);

  // = Get/set timer returned from the reactor.
  int timer_id (void);
  void timer_id (int);

  // = Get/set preemption priority.
  RtecScheduler::OS_Priority preemption_priority (void);
  void preemption_priority (RtecScheduler::OS_Priority pp);

protected:
  /// This is called when timeouts occur.  Calls correlation_->
  virtual int execute (void* arg = 0);

  /// For cancelling timers.
  int timer_id_;

  /// Store the preemption priority so we can cancel the correct timer.
  /// The priority values may change during the life.
  RtecScheduler::OS_Priority preemption_priority_;

  TAO_EC_Event timeout_event_;
};

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

/**
 * @class ACE_ES_Subscription_Info
 *
 * @brief Event Service Subscription Info
 *
 * Contains information on all consumers subscribed to a supplier.
 * Each Push_Supplier_Proxy has an instance of this class.  This
 * should really be defined in Channel_Modules.h, but I want to
 * have an instance of it in each ACE_Push_Supplier_Proxy.  This
 * allows us to reduce the amount of dynamic memory allocation.
 */
class TAO_RTOLDEvent_Export ACE_ES_Subscription_Info
{
public:
  /// Free up dynamic resources.
  ~ACE_ES_Subscription_Info (void);

  typedef ACE_Unbounded_Set_Ex_Iterator<ACE_ES_Consumer_Rep *> Subscriber_Set_Iterator;
  typedef ACE_Unbounded_Set_Ex<ACE_ES_Consumer_Rep *> Subscriber_Set;

/**
 * @class Type_Subscribers
 *
 * There is one of these for each event type generated by the
 * supplier.  It contains the subscribers and the
 * dependency_info_ describing the method that generates this
 * event type.
 */
  class Type_Subscribers
  {
  public:
    /**
     * Construction requires a dependency info describing the method
     * that generates events for the consumers_.  We use a pointer so
     * that a null can be passed in this->insert_or_allocate.
     */
    Type_Subscribers (RtecScheduler::Dependency_Info *d)
      : dependency_info_ (d) {}