Changeset 5975 for branches/newapi

Timestamp:

01/01/08 20:28:13 (11 months ago)

Author:

erijo

Message:

Simplified EventQueue?

Refactored EventQueue? into a simple queue with push/pop instead of
pushEvent/getEvent/commitEvent. The commit logic is no longer in
EventQueue?, but in some other place (yet to be decided).

Location:

branches/newapi/licq

Files:

• licq/event/eventqueue.h (modified) (4 diffs)
• src/event/eventqueue.cpp (modified) (5 diffs)
• src/event/eventqueue.h (modified) (2 diffs)
• src/event/test/eventqueuetest.cpp (modified) (2 diffs)
• src/plugin/plugininstanceimpl.cpp (modified) (1 diff)
• src/plugin/pluginmanagerimpl.cpp (modified) (1 diff)

branches/newapi/licq/licq/event/eventqueue.h

@@ -1 +1 @@
 /*
  * This file is part of Licq, an instant messaging client for UNIX.
- * Copyright (C) 2007 Erik Johansson <erijo@licq.org>
+ * Copyright (C) 2007-2008 Erik Johansson <erijo@licq.org>
  *
  * Licq is free software; you can redistribute it and/or modify
@@ -28 +28 @@
 {
 
+/**
+ * EventQueue is a thread-safe producer-consumer queue. The
+ * producer(s) pushes events to the queue with push(). The consumer
+ * notices that the file descriptor returned from getFileDescriptor()
+ * is ready for reading and reads one or more bytes. For each byte
+ * read, it calls pop(), handles the event, and then passes the event
+ * on.
+ *
+ * @threadsafe
+ */
 class EventQueue : private boost::noncopyable
 {
@@ -36 +46 @@
    * be available for reading.
    *
-   * @returns The read file descriptor.
+   * @returns The read file descriptor, configured to be non-blocking.
    */
   virtual int getFileDescriptor() const = 0;
@@ -46 +56 @@
 
   /**
-   * Pushes an event to the end of the queue. The queue takes
-   * ownership of the event and @a event is reset before the function
-   * returns.
+   * Pushes an event to the end of the queue. Once an event has been
+   * pushed to the queue, it should not be used again by the caller,
+   * since another thread may already be using it.
+   *
+   * @param event The event to add to the queue.
    */
-  virtual void pushEvent(boost::shared_ptr<Event>& event) = 0;
+  virtual void push(boost::shared_ptr<Event> event) = 0;
 
   /**
-   * @returns The event in the front of the queue.
+   * Removes the first event from the queue and returns it.
+   *
+   * @returns The first event in the queue.
    * @throws Licq::RuntimeException If the queue is empty.
    */
-  virtual boost::shared_ptr<Event> getEvent() const = 0;
-
-  /**
-   * Commit the event in front of the queue. Before committing, make
-   * sure that no reference to any Event returned from getEvent()
-   * remains. This may include calling boost::shared_ptr::reset().
-   *
-   * @throws Licq::RuntimeException If the queue is empty.
-   */
-  virtual void commitEvent() = 0;
+  virtual boost::shared_ptr<Event> pop() = 0;
 
 protected:

branches/newapi/licq/src/event/eventqueue.cpp

@@ -25 +25 @@
 #include <cassert>
 
-LicqDaemon::EventQueue::EventQueue(const std::string& name, EventManager* manager)
+LicqDaemon::EventQueue::EventQueue(const std::string& name)
   : myName(name),
-    myEventManager(manager),
     myQueueMutex(name),
     myPipe(name)
 {
-  assert(myEventManager != NULL);
-
   myPipe.setBlockingRead(false);
   myPipe.setBlockingWrite(true);
@@ -39 +36 @@
 LicqDaemon::EventQueue::~EventQueue()
 {
-  Licq::MutexLocker locker(&myQueueMutex);
-  while (!myQueue.empty())
-    commitEventInternal();
+  assert(empty());
 }
 
@@ -55 +50 @@
 }
 
-void LicqDaemon::EventQueue::pushEvent(boost::shared_ptr<Licq::Event>& event)
+void LicqDaemon::EventQueue::push(boost::shared_ptr<Licq::Event> event)
 {
-  // Take ownership of event
-  boost::shared_ptr<Licq::Event> newEvent;
-  newEvent.swap(event);
-  assert(newEvent.unique());
-
   // Add to queue
   Licq::MutexLocker locker(&myQueueMutex);
-  myQueue.push(newEvent);
+  myQueue.push(event);
 
   // Notify
@@ -70 +60 @@
 }
 
-boost::shared_ptr<Licq::Event> LicqDaemon::EventQueue::getEvent() const
+boost::shared_ptr<Licq::Event> LicqDaemon::EventQueue::pop()
 {
   Licq::MutexLocker locker(&myQueueMutex);
@@ -76 +66 @@
     throw Licq::RuntimeException(tr("EventQueue %1% is empty") % myName);
 
-  return myQueue.front();
-}
-
-void LicqDaemon::EventQueue::commitEvent()
-{
-  Licq::MutexLocker locker(&myQueueMutex);
-  if (myQueue.empty())
-    throw Licq::RuntimeException(tr("EventQueue %1% is empty") % myName);
-
-  commitEventInternal();
-}
-
-void LicqDaemon::EventQueue::commitEventInternal()
-{
   boost::shared_ptr<Licq::Event> event = myQueue.front();
   myQueue.pop();
-
-  myEventManager->sendEvent(event);
+  return event;
 }

branches/newapi/licq/src/event/eventqueue.h

@@ -23 +23 @@
 #include "licq/event/eventqueue.h"
 #include "licq/thread/mutex.h"
-#include "event/eventmanager.h"
 #include "util/pipe.h"
 
@@ -34 +33 @@
 {
 public:
-  EventQueue(const std::string& name, EventManager* manager);
+  EventQueue(const std::string& name);
   ~EventQueue();
 
   // From Licq::EventQueue
   int getFileDescriptor() const;
-  void pushEvent(boost::shared_ptr<Licq::Event>& event);
   bool empty() const;
-  boost::shared_ptr<Licq::Event> getEvent() const;
-  void commitEvent();
+  void push(boost::shared_ptr<Licq::Event> event);
+  boost::shared_ptr<Licq::Event> pop();
 
 private:
-  void commitEventInternal();
+  typedef std::queue< boost::shared_ptr<Licq::Event> > Queue;
+  Queue myQueue;
 
   const std::string myName;
-  EventManager* myEventManager;
-
-  typedef std::queue< boost::shared_ptr<Licq::Event> > Queue;
-  Queue myQueue;
   mutable Licq::Mutex myQueueMutex;
-
   Pipe myPipe;
 };

branches/newapi/licq/src/event/test/eventqueuetest.cpp

@@ -18 +18 @@
  */
 
-#include "event/eventmanager.h"
 #include "event/eventqueue.h"
-#include "test/recorder.h"
+#include "licq/exception/runtimeexception.h"
 
 #include <boost/test/unit_test.hpp>
-#include <cassert>
 
-struct EventQueueFixture : public LicqDaemon::EventManager
+struct EventQueueFixture
 {
-  void reset()
+  ssize_t read(Licq::EventQueue* queue)
   {
-    BOOST_REQUIRE(recorder.isEmpty());
+    int fd = queue->getFileDescriptor();
+
+    char buffer[16];
+    return ::read(fd, &buffer, sizeof(buffer));
   }
-
-  // EventManager
-  void registerQueue(const std::string&, Licq::EventQueue*, int)
-  {
-    assert(false);
-  }
-
-  void unregisterQueue(const std::string&, Licq::EventQueue*)
-  {
-    assert(false);
-  }
-
-  void unregisterQueue(Licq::EventQueue*)
-  {
-    assert(false);
-  }
-
-  void sendEvent(boost::shared_ptr<Licq::Event>& event)
-  {
-    recorder.push(boost::format("sendEvent(%1%)") % event->getId());
-  }
-
-  LicqTest::Recorder recorder;
 };
 
@@ -60 +38 @@
 BOOST_AUTO_TEST_CASE(basic)
 {
-  reset();
+  const size_t count = 5;
+  boost::shared_ptr<Licq::Event> event[count];
+  Licq::Event::Id id[count];
 
-  LicqDaemon::EventQueue queue("test", this);
-  boost::shared_ptr<Licq::Event> test = Licq::Event::create("test");
-  Licq::Event* testPtr = test.get();
-  Licq::Event::Id id = test->getId();
+  // Setup some events
+  for (size_t i = 0; i < count; i++)
+  {
+    event[i] = Licq::Event::create("test");
+    id[i] = event[i]->getId();
+    event[i]->setProperty("42 * i", (int)(42 * i));
+  }
 
-  queue.pushEvent(test);
-  BOOST_CHECK_EQUAL(queue.empty(), false);
+  boost::shared_ptr<Licq::Event> temp;
 
-  // Adding an event should invalidate the pointer
-  BOOST_CHECK_EQUAL(test.get(), (Licq::Event*)NULL);
+  LicqDaemon::EventQueue queue("testqueue");
+  BOOST_CHECK(queue.empty());
 
-  BOOST_CHECK_EQUAL(queue.getEvent().get(), testPtr);
-  queue.commitEvent();
-  BOOST_CHECK_EQUAL(queue.empty(), true);
+  queue.push(event[0]);
+  BOOST_CHECK(!queue.empty());
+  BOOST_CHECK_EQUAL(read(&queue), 1);
 
-  BOOST_CHECK(recorder.pop(boost::format("sendEvent(%1%)") % id));
-  BOOST_CHECK(recorder.isEmpty());
+  // pop() should not leave the event in the queue
+  temp = queue.pop();
+  BOOST_CHECK_EQUAL(temp, event[0]);
+  BOOST_CHECK_EQUAL(temp.use_count(), 2);
+
+  // Push events in order
+  for (size_t i = 0; i < count; i++)
+    queue.push(event[i]);
+
+  BOOST_CHECK_EQUAL(read(&queue), count);
+
+  // The queue should maintain order
+  for (size_t i = 0; i < count; i++)
+  {
+    temp = queue.pop();
+    BOOST_CHECK_EQUAL(temp, event[i]);
+    BOOST_CHECK_EQUAL(temp->getId(), id[i]);
+    BOOST_CHECK_EQUAL(temp->getProperty<int>("42 * i"), (int)(42 * i));
+  }
+
+  BOOST_CHECK(queue.empty());
+  BOOST_CHECK_THROW(queue.pop(), Licq::RuntimeException);
 }
 

branches/newapi/licq/src/plugin/plugininstanceimpl.cpp

@@ -94 +94 @@
 
   // Create the plugin's event queue
-  mySetup.eventQueue = new EventQueue(myInformation->name,
-                                      Globals::getEventManager());
+  mySetup.eventQueue = new EventQueue(myInformation->name);
 
   // Create the plugin

branches/newapi/licq/src/plugin/pluginmanagerimpl.cpp

@@ -68 +68 @@
   // Send started event
   boost::shared_ptr<Licq::Event> started = Licq::Event::create("started");
-  instance->getEventQueue()->pushEvent(started);
+  instance->getEventQueue()->push(started);
 
   return id;