Initial revision

1 files changed, 512 insertions, 0 deletions

diff --git a/ace/Stream.cpp b/ace/Stream.cpp
new file mode 100644
index 00000000000..7b17d555b80
--- /dev/null
+++ b/ace/Stream.cpp
@@ -0,0 +1,512 @@
+// Stream.cpp
+// $Id$
+
+#if !defined (ACE_STREAM_C)
+#define ACE_STREAM_C
+
+#define ACE_BUILD_DLL
+//#include "ace/Module.h"
+#include "ace/Stream.h"
+#include "ace/Stream_Modules.h"
+
+#if !defined (__ACE_INLINE__)
+#include "ace/Stream.i"
+#endif /* __ACE_INLINE__ */
+
+ACE_ALLOC_HOOK_DEFINE(ACE_Stream)
+
+// Give some idea of what the heck is going on in a stream!
+
+template <ACE_SYNCH_1> void
+ACE_Stream<ACE_SYNCH_2>::dump (void) const
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::dump");
+  ACE_DEBUG ((LM_DEBUG, "-------- module links --------\n"));
+
+  for (ACE_Module<ACE_SYNCH_2> *mp = this->stream_head_; ; mp = mp->next ())
+    {
+      ACE_DEBUG ((LM_DEBUG, "module name = %s\n", mp->name ()));
+      if (mp == this->stream_tail_)
+	break;
+    }
+
+  ACE_DEBUG ((LM_DEBUG, "-------- writer links --------\n"));
+
+  ACE_Task<ACE_SYNCH_2> *qp;
+
+  for (qp = this->stream_head_->writer (); ; qp = qp->next ())
+    {
+      ACE_DEBUG ((LM_DEBUG, "writer queue name = %s\n", qp->name ()));
+      qp->dump ();
+      ACE_DEBUG ((LM_DEBUG, "-------\n"));
+      if (qp == this->stream_tail_->writer ()
+	  || (this->linked_us_ && qp == this->linked_us_->stream_head_->reader ()))
+	break;
+    }
+
+  ACE_DEBUG ((LM_DEBUG, "-------- reader links --------\n"));
+  for (qp = this->stream_tail_->reader (); ; qp = qp->next ())
+    {
+      ACE_DEBUG ((LM_DEBUG, "reader queue name = %s\n", qp->name ()));
+      qp->dump ();
+      ACE_DEBUG ((LM_DEBUG, "-------\n"));
+      if (qp == this->stream_head_->reader ()
+	  || (this->linked_us_ && qp == this->linked_us_->stream_head_->writer ()))
+	break;
+    }
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::push (ACE_Module<ACE_SYNCH_2> *new_top)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::push");
+  if (this->push_module  (new_top, 
+			  this->stream_head_->next (), 
+			  this->stream_head_) == -1)
+    return -1;
+  else
+    return 0;
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream_Iterator<ACE_SYNCH_2>::advance (void)
+{
+  ACE_TRACE ("ACE_Stream_Iterator<ACE_SYNCH_2>::advance");
+  this->next_ = this->next_->next ();
+  return 0;
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::put (ACE_Message_Block *mb, ACE_Time_Value *tv)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::put");
+  return this->stream_head_->writer ()->put (mb, tv);
+}
+
+template <ACE_SYNCH_1> int 
+ACE_Stream<ACE_SYNCH_2>::get (ACE_Message_Block *&mb, ACE_Time_Value *tv)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::get");
+  return this->stream_head_->reader ()->getq (mb, tv);
+}
+
+// Return the "top" ACE_Module in a ACE_Stream, skipping over the
+// stream_head.
+
+template <ACE_SYNCH_1> int 
+ACE_Stream<ACE_SYNCH_2>::top (ACE_Module<ACE_SYNCH_2> *&m)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::top");
+  if (this->stream_head_->next () == this->stream_tail_)
+    return -1;
+  else
+    {
+      m = this->stream_head_->next ();
+      return 0;
+    }
+}
+
+// Remove the "top" ACE_Module in a ACE_Stream, skipping over the
+// stream_head.
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::pop (u_long flags)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::pop");
+  if (this->stream_head_->next () == this->stream_tail_)
+    return -1;
+  else
+    {
+      // Skip over the ACE_Stream head.
+      ACE_Module<ACE_SYNCH_2> *top = this->stream_head_->next ();
+      ACE_Module<ACE_SYNCH_2> *new_top = top->next ();
+
+      this->stream_head_->next (new_top);
+
+      // Close the top ACE_Module.
+
+      top->close (flags);
+
+      this->stream_head_->writer ()->next (new_top->writer ());
+      new_top->reader ()->next (this->stream_head_->reader ());
+
+      return 0;
+    }
+}
+
+// Remove a named ACE_Module from an arbitrary place in the
+// ACE_Stream.
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::remove (const char *name, u_long flags)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::remove");
+  ACE_Module<ACE_SYNCH_2> *prev = 0;
+
+  for (ACE_Module<ACE_SYNCH_2> *mod = this->stream_head_; mod != 0; mod = mod->next ())
+    if (ACE_OS::strcmp (mod->name (), name) == 0)
+      {
+	if (prev == 0) // Deleting ACE_Stream Head
+	  this->stream_head_->link (mod->next ());
+	else
+	  prev->link (mod->next ());
+
+	mod->close (flags);
+	return 0;
+      }
+    else
+      prev = mod;
+
+  return -1;
+}
+
+template <ACE_SYNCH_1> ACE_Module<ACE_SYNCH_2> *
+ACE_Stream<ACE_SYNCH_2>::find (const char *name)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::find");
+  for (ACE_Module<ACE_SYNCH_2> *mod = this->stream_head_; 
+       mod != 0; 
+       mod = mod->next ())
+    if (ACE_OS::strcmp (mod->name (), name) == 0)
+	return mod;
+
+  return 0;
+}
+
+// Actually push a module onto the stack...
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::push_module (ACE_Module<ACE_SYNCH_2> *new_top, 
+				      ACE_Module<ACE_SYNCH_2> *current_top, 
+				      ACE_Module<ACE_SYNCH_2> *head)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::push_module");
+  ACE_Task<ACE_SYNCH_2> *nt_reader = new_top->reader ();
+  ACE_Task<ACE_SYNCH_2> *nt_writer = new_top->writer ();
+  ACE_Task<ACE_SYNCH_2> *ct_reader = 0;
+  ACE_Task<ACE_SYNCH_2> *ct_writer = 0;
+
+  if (current_top)
+    {
+      ct_reader = current_top->reader ();
+      ct_writer = current_top->writer ();
+      ct_reader->next (nt_reader);
+    }
+
+  nt_writer->next (ct_writer);
+
+  if (head)
+    {
+      if (head != new_top)
+	head->link (new_top);
+    }
+  else
+    nt_reader->next (0);
+
+  new_top->next (current_top);
+
+  if (nt_reader->open (new_top->arg ()) == -1)
+    return -1;
+
+  if (nt_writer->open (new_top->arg ()) == -1)
+    return -1;
+  return 0;
+}
+
+#if 0
+int
+ACE_Stream<ACE_SYNCH_2>::open (void *a, 
+			   ACE_Multiplexor &muxer, 
+			   ACE_Module<ACE_SYNCH_2> *head)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::open");
+  this->stream_head_ = head == 0
+    ? new ACE_Module<ACE_SYNCH_2> ("ACE_Stream_Head", 
+			       new ACE_Stream_Head<ACE_SYNCH_2>, 
+			       new ACE_Stream_Head<ACE_SYNCH_2>, a) : head;
+  this->stream_tail_ = 0;
+  return muxer.link_from_below (this->stream_head_);
+}
+#endif 
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::open (void *a, 
+			       ACE_Module<ACE_SYNCH_2> *head, 
+			       ACE_Module<ACE_SYNCH_2> *tail)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::open");
+  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
+  ACE_Task<ACE_SYNCH_2> *h1, *h2;
+  ACE_Task<ACE_SYNCH_2> *t1, *t2;
+
+  if (head == 0)
+    {
+      h1 = new ACE_Stream_Head<ACE_SYNCH_2>;
+      h2 = new ACE_Stream_Head<ACE_SYNCH_2>;
+      head = new ACE_Module<ACE_SYNCH_2> ("ACE_Stream_Head", h1, h2, a);
+    }
+
+  if (tail == 0)
+    {
+      t1 = new ACE_Stream_Tail<ACE_SYNCH_2>;
+      t2 = new ACE_Stream_Tail<ACE_SYNCH_2>;
+      tail = new ACE_Module<ACE_SYNCH_2> ("ACE_Stream_Tail", 
+					  t1, t2, a);
+    }
+
+  // Make sure *all* the allocation succeeded!
+  if (h1 == 0 || h2 == 0 || head == 0
+      || t1 == 0 || t2 == 0 || tail == 0)
+    {
+      delete h1;
+      delete h2;
+      delete t1;
+      delete t2;
+      // Note that we can't call delete on these, we must call close!
+      head->close ();
+      tail->close ();
+      errno = ENOMEM;
+      return -1;
+    }
+
+  this->stream_head_ = head;
+  this->stream_tail_ = tail;
+
+  if (this->push_module (this->stream_tail_) == -1)
+    return -1;
+  else if (this->push_module (this->stream_head_, 
+			      this->stream_tail_, 
+			      this->stream_head_) == -1)
+    return -1;
+  else
+    return 0;
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::close (u_long flags)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::close");
+  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
+
+  if (this->stream_head_ != 0 
+      && this->stream_tail_ != 0)
+    {
+      // Don't bother checking return value here.
+      this->unlink_i ();
+
+      int result = 0;
+
+      // Remove and cleanup all the intermediate modules.
+
+      while (this->stream_head_->next () != this->stream_tail_)
+	{
+	  if (this->pop (flags) == -1)
+	    result = -1;
+	}
+  
+      // Clean up the head and tail of the stream.
+      if (this->stream_head_->close (flags) == -1)
+	result = -1;
+      if (this->stream_tail_->close (flags) == -1)
+	result = -1;
+
+      this->stream_head_ = 0;
+      this->stream_tail_ = 0;
+
+      // Tell all threads waiting on the close that we are done.
+      this->final_close_.broadcast ();
+      return result;
+    }
+  return 0;
+}
+
+template <ACE_SYNCH_1> int 
+ACE_Stream<ACE_SYNCH_2>::control (ACE_IO_Cntl_Msg::ACE_IO_Cntl_Cmds cmd, 
+				  void *a)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::control");
+  ACE_IO_Cntl_Msg ioc (cmd);
+
+  // Create a data block that contains the user-supplied data.
+  ACE_Message_Block *db = 
+    new ACE_Message_Block (sizeof (int), 
+			   ACE_Message_Block::MB_IOCTL, 
+			   0, 
+			   (char *) a);
+
+  // Create a control block that contains the control field and a
+  // pointer to the data block.
+  ACE_Message_Block *cb = 
+    new ACE_Message_Block (sizeof ioc, 
+			   ACE_Message_Block::MB_IOCTL, 
+			   db, 
+			   (char *) &ioc);
+
+  // Make sure all of the allocation succeeded before continuing.
+  if (db == 0 || cb == 0)
+    {
+      delete cb; 
+      delete db; 
+      errno = ENOMEM;
+      return -1;
+    }
+
+  int result = 0;
+
+  if (this->stream_head_->writer ()->put (cb) == -1)
+    result = -1;
+  else if (this->stream_head_->reader ()->getq (cb) == -1)
+    result = -1;
+  else
+    result = ((ACE_IO_Cntl_Msg *) cb->rd_ptr ())->rval ();
+
+  delete cb; // This also deletes db...
+  return result;
+}
+
+// Link two streams together at their bottom-most Modules (i.e., the
+// one just above the Stream tail).  Note that all of this is premised
+// on the fact that the Stream head and Stream tail are non-NULL...
+// This must be called with locks held.
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::link_i (ACE_Stream<ACE_SYNCH_2> &us)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::link_i");
+  this->linked_us_ = &us;
+  // Make sure the other side is also linked to us!
+  us.linked_us_ = this;
+
+  ACE_Module<ACE_SYNCH_2> *my_tail = this->stream_head_;
+
+  if (my_tail == 0)
+    return -1;
+
+  // Locate the module just above our Stream tail. 
+  while (my_tail->next () != this->stream_tail_)
+    my_tail = my_tail->next ();
+
+  ACE_Module<ACE_SYNCH_2> *other_tail = us.stream_head_;
+
+  if (other_tail == 0)
+    return -1;
+
+  // Locate the module just above the other Stream's tail.
+  while (other_tail->next () != us.stream_tail_)
+    other_tail = other_tail->next ();
+
+  // Reattach the pointers so that the two streams are linked!
+  my_tail->writer ()->next (other_tail->reader ());
+  other_tail->writer ()->next (my_tail->reader ());
+  return 0;
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::link (ACE_Stream<ACE_SYNCH_2> &us)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::link");
+  
+  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
+
+  return this->link_i (us);
+}  
+
+// Must be called with locks held...
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::unlink_i (void)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::unlink_i");
+
+  // Only try to unlink if we are in fact still linked!
+
+  if (this->linked_us_ != 0)
+    {
+      ACE_Module<ACE_SYNCH_2> *my_tail = this->stream_head_;
+
+      // Only relink if we still exist!
+      if (my_tail)
+	{
+	  // Find the module that's just before our stream tail.
+	  while (my_tail->next () != this->stream_tail_)
+	    my_tail = my_tail->next ();
+
+	  // Restore the writer's next() link to our tail.
+	  my_tail->writer ()->next (this->stream_tail_->writer ());
+	}
+
+      ACE_Module<ACE_SYNCH_2> *other_tail = 
+	this->linked_us_->stream_head_;
+
+      // Only fiddle with the other side if it in fact still remains.
+      if (other_tail != 0)
+	{
+	  while (other_tail->next () != this->linked_us_->stream_tail_)
+	    other_tail = other_tail->next ();
+
+	  other_tail->writer ()->next (this->linked_us_->stream_tail_->writer ());
+
+	}
+
+      // Make sure the other side is also aware that it's been unlinked!
+      this->linked_us_->linked_us_ = 0;
+
+      this->linked_us_ = 0;
+      return 0;
+    }
+  else
+    return -1;
+}
+
+template <ACE_SYNCH_1> int
+ACE_Stream<ACE_SYNCH_2>::unlink (void)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::unlink");
+  ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, ace_mon, this->lock_, -1);
+  return this->unlink_i ();
+}
+
+template <ACE_SYNCH_1> ACE_INLINE
+ACE_Stream<ACE_SYNCH_2>::ACE_Stream (void * a, 
+				     ACE_Module<ACE_SYNCH_2> *head, 
+				     ACE_Module<ACE_SYNCH_2> *tail)
+  : final_close_ (this->lock_), 
+    linked_us_ (0)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::ACE_Stream");
+  if (this->open (a, head, tail) == -1)
+    ACE_ERROR ((LM_ERROR, "ACE_Stream<ACE_SYNCH_2>::open (%s, %s)\n", 
+	       head->name (), tail->name ()));
+}
+
+#if 0
+ACE_INLINE
+ACE_Stream<ACE_SYNCH_2>::ACE_Stream (void *a, 
+				     ACE_Multiplexor &muxer, 
+				     ACE_Module<ACE_SYNCH_2> *head)
+  : final_close_ (this->lock_), 
+    linked_us_ (0)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::ACE_Stream");
+  if (this->open (a, muxer, head) == -1)
+    ACE_ERROR ((LM_ERROR, "ACE_Stream<ACE_SYNCH_2>::open (%s, %s)\n", 
+		head->name ()));
+}
+#endif 
+
+template <ACE_SYNCH_1> ACE_INLINE
+ACE_Stream<ACE_SYNCH_2>::~ACE_Stream (void)
+{
+  ACE_TRACE ("ACE_Stream<ACE_SYNCH_2>::~ACE_Stream");
+  if (this->stream_head_ != 0)
+    this->close ();
+}
+
+template <ACE_SYNCH_1> ACE_INLINE
+ACE_Stream_Iterator<ACE_SYNCH_2>::ACE_Stream_Iterator (const ACE_Stream<ACE_SYNCH_2> &sr)
+  : next_ (sr.stream_head_)
+{
+  ACE_TRACE ("ACE_Stream_Iterator<ACE_SYNCH_2>::ACE_Stream_Iterator");
+}
+
+#endif /* ACE_STREAM_C */