ACE/ace/Stream.h

   1 // -*- C++ -*-
   2
   3 //==========================================================================
   4 /**
   5  *  @file    Stream.h
   6  *
   7  *  @author Douglas C. Schmidt <d.schmidt@vanderbilt.edu>
   8  */
   9 //==========================================================================
  10
  11 #ifndef ACE_STREAM_H
  12 #define ACE_STREAM_H
  13
  14 #include /**/ "ace/pre.h"
  15
  16 #include /**/ "ace/config-all.h"
  17
  18 #if !defined (ACE_LACKS_PRAGMA_ONCE)
  19 # pragma once
  20 #endif /* ACE_LACKS_PRAGMA_ONCE */
  21
  22 #include "ace/IO_Cntl_Msg.h"
  23 #include "ace/Message_Block.h"
  24 #include "ace/Module.h"
  25 #if defined (ACE_HAS_THREADS)
  26 # include "ace/Condition_Attributes.h"
  27 #endif
  28
  29 ACE_BEGIN_VERSIONED_NAMESPACE_DECL
  30
  31 // Forward decls.
  32 template<ACE_SYNCH_DECL, class TIME_POLICY> class ACE_Stream_Iterator;
  33 class ACE_Time_Value;
  34
  35 /**
  36  * @class ACE_Stream
  37  *
  38  * @brief This class is the primary abstraction for the ASX framework.
  39  * It is moduled after System V Stream.
  40  *
  41  * A Stream consists of a stack of @c ACE_Modules, each of which
  42  * contains two @c ACE_Tasks.  Even though the methods in this
  43  * class are virtual, this class isn't really intended for
  44  * subclassing unless you know what you are doing.  In
  45  * particular, the ACE_Stream destructor calls <close>, which
  46  * won't be overridden properly unless you call it in a subclass
  47  * destructor.
  48  */
  49 template <ACE_SYNCH_DECL, class TIME_POLICY = ACE_System_Time_Policy>
  50 class ACE_Stream
  51 {
  52 public:
  53   friend class ACE_Stream_Iterator<ACE_SYNCH_USE, TIME_POLICY>;
  54
  55   enum
  56   {
  57     /// Indicates that @c close() deletes the Tasks.  Don't change this
  58     /// value without updating the same enum in class ACE_Module...
  59     M_DELETE = 3
  60   };
  61
  62   // = Initializatation and termination methods.
  63   /**
  64    * Create a Stream consisting of @a head and @a tail as the Stream
  65    * head and Stream tail, respectively.  If these are 0 then the
  66    * ACE_Stream_Head and ACE_Stream_Tail are used, respectively.
  67    * @a arg is the value past in to the <open> methods of the tasks.
  68    */
  69   ACE_Stream (void *arg = 0,
  70               ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *head = 0,
  71               ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *tail = 0);
  72
  73   /**
  74    * Create a Stream consisting of @a head and @a tail as the Stream
  75    * head and Stream tail, respectively.  If these are 0 then the
  76    * ACE_Stream_Head and ACE_Stream_Tail are used, respectively.
  77    * @a arg is the value past in to the @c open() methods of the tasks.
  78    */
  79   virtual int open (void *arg,
  80                     ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *head = 0,
  81                     ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *tail = 0);
  82
  83   /// Close down the stream and release all the resources.
  84   virtual int close (int flags = M_DELETE);
  85
  86   /// Close down the stream and release all the resources.
  87   virtual ~ACE_Stream ();
  88
  89   // = ACE_Stream plumbing operations
  90
  91   /// Add a new module @a mod right below the Stream head.  The
  92   /// @c open() hook methods of the @c ACE_Tasks in this ACE_Module
  93   /// are invoked to initialize the tasks.
  94   virtual int push (ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *mod);
  95
  96   /// Remove the @a mod right below the Stream head and close it down.
  97   //  The <close()> hook methods of the <ACE_Tasks> in this ACE_Module
  98   /// are invoked to cleanup the tasks.
  99   virtual int pop (int flags = M_DELETE);
 100
 101   /// Return the top module on the stream (right below the stream
 102   /// head).
 103   virtual int top (ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *&mod);
 104
 105   /// Insert a new module @a mod below the named module @a prev_name.
 106   virtual int insert (const ACE_TCHAR *prev_name,
 107                       ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *mod);
 108
 109   /// Replace the named module @a replace_name with a new module @a mod.
 110   virtual int replace (const ACE_TCHAR *replace_name,
 111                        ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *mod,
 112                        int flags = M_DELETE);
 113
 114   /// Remove the named module @a mod from the stream.  This bypasses the
 115   /// strict LIFO ordering of @c push and @c pop.
 116   virtual int remove (const ACE_TCHAR *mod,
 117                       int flags = M_DELETE);
 118
 119   /// Return current stream head.
 120   virtual ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *head ();
 121
 122   /// Return current stream tail.
 123   virtual ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *tail ();
 124
 125   /// Find a particular ACE_Module.
 126   virtual ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *find (const ACE_TCHAR *mod);
 127
 128   /// Create a pipe between two Streams.
 129   virtual int link (ACE_Stream<ACE_SYNCH_USE, TIME_POLICY> &);
 130
 131   /// Remove a pipe formed between two Streams.
 132   virtual int unlink ();
 133
 134   // = Blocking data transfer operations
 135   /**
 136    * Send the message @a mb down the stream, starting at the Module
 137    * below the Stream head.  Wait for upto @a timeout amount of
 138    * absolute time for the operation to complete (or block forever if
 139    * @a timeout == 0).
 140    */
 141   virtual int put (ACE_Message_Block *mb,
 142                    ACE_Time_Value *timeout = 0);
 143
 144   /**
 145    * Read the message @a mb that is stored in the stream head.
 146    * Wait for upto @a timeout amount of absolute time for the operation
 147    * to complete (or block forever if @a timeout == 0).
 148    */
 149   virtual int get (ACE_Message_Block *&mb,
 150                    ACE_Time_Value *timeout = 0);
 151
 152   /// Send control message down the stream.
 153   virtual int control (ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds cmd,
 154                        void *args);
 155
 156   /// Synchronize with the final close of the stream.
 157   virtual int wait ();
 158
 159   /// Dump the state of an object.
 160   virtual void dump () const;
 161
 162   /// Declare the dynamic allocation hooks.
 163   ACE_ALLOC_HOOK_DECLARE;
 164
 165 protected:
 166   /// Pointer to the head of the stream.
 167   ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *stream_head_;
 168
 169   /// Pointer to the tail of the stream.
 170   ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *stream_tail_;
 171
 172   /// Pointer to an adjoining linked stream.
 173   ACE_Stream<ACE_SYNCH_USE, TIME_POLICY> *linked_us_;
 174
 175   // = Synchronization objects used for thread-safe streams.
 176   /// Protect the stream against race conditions.
 177   mutable ACE_SYNCH_MUTEX_T lock_;
 178
 179 #if defined (ACE_HAS_THREADS)
 180   /// Attributes to initialize condition with.
 181   /* We only need this because some crappy compilers can't
 182      properly handle initializing the conditions with
 183      temporary objects. */
 184   ACE_Condition_Attributes_T<TIME_POLICY> cond_attr_;
 185 #endif
 186
 187   /// Use to tell all threads waiting on the close that we are done.
 188   ACE_SYNCH_CONDITION_T final_close_;
 189
 190 private:
 191   /// Actually perform the unlinking of two Streams (must be called
 192   /// with locks held).
 193   int unlink_i ();
 194
 195   /// Actually perform the linking of two Streams (must be called with
 196   /// locks held).
 197   int link_i (ACE_Stream<ACE_SYNCH_USE, TIME_POLICY> &);
 198
 199   /// Must a new module onto the Stream.
 200   int push_module (ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *,
 201                    ACE_Module<ACE_SYNCH_USE, TIME_POLICY> * = 0,
 202                    ACE_Module<ACE_SYNCH_USE, TIME_POLICY> * = 0);
 203 };
 204
 205 /**
 206  * @class ACE_Stream_Iterator
 207  *
 208  * @brief Iterate through an ACE_Stream.
 209  */
 210 template <ACE_SYNCH_DECL, class TIME_POLICY = ACE_System_Time_Policy>
 211 class ACE_Stream_Iterator
 212 {
 213 public:
 214   ACE_Stream_Iterator (const ACE_Stream<ACE_SYNCH_USE, TIME_POLICY> &sr);
 215
 216   // = Iteration methods.
 217
 218   /// Pass back the @a next_item that hasn't been seen in the set.
 219   /// Returns 0 when all items have been seen, else 1.
 220   int next (const ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *&next_item);
 221
 222   /// Returns 1 when all items have been seen, else 0.
 223   int done () const;
 224
 225   /// Move forward by one element in the set.  Returns 0 when all the
 226   /// items in the set have been seen, else 1.
 227   int advance ();
 228
 229 private:
 230   /// Next ACE_Module that we haven't yet seen.
 231   ACE_Module<ACE_SYNCH_USE, TIME_POLICY> *next_;
 232 };
 233
 234 ACE_END_VERSIONED_NAMESPACE_DECL
 235
 236 #if defined (__ACE_INLINE__)
 237 #include "ace/Stream.inl"
 238 #endif /* __ACE_INLINE__ */
 239
 240 #include "ace/Stream.cpp"
 241
 242 #include /**/ "ace/post.h"
 243
 244 #endif /* ACE_STREAM_H */