Merge pull request #2011 from esohns/message_queue_ex_get_queue

ACE_Message_Queue_Ex: support accessing the underlying queue (e.g. for iteration)

3 files changed, 117 insertions, 0 deletions

diff --git a/ACE/ace/Message_Queue_T.cpp b/ACE/ace/Message_Queue_T.cpp
index 76ff7aa38fa..6ff89561fa5 100644
--- a/ACE/ace/Message_Queue_T.cpp
+++ b/ACE/ace/Message_Queue_T.cpp
@@ -691,6 +691,13 @@ ACE_Message_Queue_Ex<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::set_time_pol
   this->queue_.set_time_policy (rhs);
 }
 
+template <class ACE_MESSAGE_TYPE, ACE_SYNCH_DECL, class TIME_POLICY>
+ACE_Message_Queue<ACE_SYNCH_USE, TIME_POLICY> &
+ACE_Message_Queue_Ex<ACE_MESSAGE_TYPE, ACE_SYNCH_USE, TIME_POLICY>::queue ()
+{
+  return this->queue_;
+}
+
 template <ACE_SYNCH_DECL, class TIME_POLICY>
 ACE_Message_Queue_Iterator<ACE_SYNCH_USE, TIME_POLICY>::ACE_Message_Queue_Iterator (ACE_Message_Queue <ACE_SYNCH_USE, TIME_POLICY> &q)
   : queue_ (q)
diff --git a/ACE/ace/Message_Queue_T.h b/ACE/ace/Message_Queue_T.h
index cfd3fdd43a6..f4be9145dde 100644
--- a/ACE/ace/Message_Queue_T.h
+++ b/ACE/ace/Message_Queue_T.h
@@ -1385,6 +1385,11 @@ public:
   /// Dump the state of an object.
   virtual void dump () const;
 
+  /// Support access to the underlying <Message_Queue>. Note that
+  /// manipulating the lower level queue directly may be hazardous (, but
+  /// necessary in some scenarios); be sure to lock the queue first.
+  ACE_Message_Queue<ACE_SYNCH_USE, TIME_POLICY> &queue ();
+
   /// Declare the dynamic allocation hooks.
   ACE_ALLOC_HOOK_DECLARE;
 
diff --git a/ACE/tests/Message_Queue_Test_Ex.cpp b/ACE/tests/Message_Queue_Test_Ex.cpp
index 853eebb8504..6b698629a1b 100644
--- a/ACE/tests/Message_Queue_Test_Ex.cpp
+++ b/ACE/tests/Message_Queue_Test_Ex.cpp
@@ -709,6 +709,105 @@ int queue_priority_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
   return status;
 }
 
+class Queue_Ex_Iterator_No_Lock
+ : public ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>
+{
+ public:
+  typedef ACE_Message_Queue_Ex<User_Class, ACE_SYNCH, ACE_System_Time_Policy> MESSAGE_QUEUE_EX_T;
+
+  explicit Queue_Ex_Iterator_No_Lock (MESSAGE_QUEUE_EX_T& queue_in)
+    : ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy> (queue_in.queue ())
+  {}
+  virtual ~Queue_Ex_Iterator_No_Lock () = default;
+
+  int next (User_Class*& message_inout)
+  {
+    if (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_)
+    {
+      message_inout =
+            reinterpret_cast<User_Class*> (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_->base ());
+      return 1;
+    }
+
+    return 0;
+  }
+
+  int done () const
+  {
+    return (!ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ ? 1 : 0);
+  }
+
+  int advance ()
+  {
+    if (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_)
+    {
+      ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ =
+        ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_->next ();
+    }
+
+    return (ACE_Message_Queue_Iterator<ACE_SYNCH, ACE_System_Time_Policy>::curr_ ? 1 : 0);
+  }
+};
+
+int queue_iterator_test (ACE_Message_Queue_Ex<User_Class, ACE_SYNCH>& q)
+{
+  int status = 0;
+  if (!q.is_empty ())
+    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("Iterator test queue not empty\n")), 1);
+
+  // Set up a few objects with names for how they should come out of the queue.
+  std::unique_ptr<User_Class> b1 = std::make_unique<User_Class> ("first");
+  std::unique_ptr<User_Class> b2 = std::make_unique<User_Class> ("second");
+  std::unique_ptr<User_Class> b3 = std::make_unique<User_Class> ("third");
+  std::unique_ptr<User_Class> b4 = std::make_unique<User_Class> ("fourth");
+  if (-1 == q.enqueue_tail (b1.get (), 0))
+    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b1")), 1);
+  if (-1 == q.enqueue_tail (b2.get (), 0))
+    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b2")), 1);
+  if (-1 == q.enqueue_tail (b3.get (), 0))
+    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b3")), 1);
+  if (-1 == q.enqueue_tail (b4.get (), 0))
+    ACE_ERROR_RETURN ((LM_ERROR, ACE_TEXT ("%p\n"), ACE_TEXT ("b4")), 1);
+
+  User_Class* b = nullptr;
+  {
+    ACE_GUARD_RETURN (ACE_SYNCH_MUTEX, aGuard, q.lock (), 1);
+    int counter = 0;
+    for (Queue_Ex_Iterator_No_Lock iterator (q);
+         iterator.next (b);
+         iterator.advance ())
+    { 
+      ACE_ASSERT (b);
+      ++counter;
+      if (counter == 1)
+      {
+        if (ACE_OS::strcmp (b->message (), "first") != 0)
+        {
+          ACE_ERROR ((LM_ERROR, ACE_TEXT ("First message was %C\n"), b->message ()));
+          ++status;
+        }
+      }
+      else if (counter == 4)
+      {
+        if (ACE_OS::strcmp (b->message (), "fourth") != 0)
+        {
+          ACE_ERROR ((LM_ERROR, ACE_TEXT ("Fourth message was %C\n"), b->message ()));
+          ++status;
+        }
+      }
+
+      b = nullptr;
+    }
+  }
+
+  while (!q.is_empty ())
+    q.dequeue_head (b, 0);
+
+  if (status == 0)
+    ACE_DEBUG ((LM_DEBUG, ACE_TEXT ("Iterator test: OK\n")));
+  return status;
+}
+
 int
 run_main (int argc, ACE_TCHAR *argv[])
 {
@@ -745,6 +844,12 @@ run_main (int argc, ACE_TCHAR *argv[])
       ++status;
     }
 
+  // Check iterator operations.
+  if (0 != queue_iterator_test (q1))
+    {
+      ++status;
+    }
+
   ACE_NEW_RETURN (timer,
                   ACE_High_Res_Timer,
                   -1);