Changeset 304

Timestamp:

02/25/08 06:28:26 (9 months ago)

Author:

rm

Message:

more work on SpreadConnection?, getting it working. initial introduction of unit test infrastructure, cppunit.

Location:

trunk/thrucommon

Files:

• Makefile.am (modified) (1 diff)
• configure.in (modified) (2 diffs)
• src/Makefile.am (modified) (2 diffs)
• src/SpreadConnection.cpp (modified) (15 diffs)
• src/SpreadConnection.h (modified) (4 diffs)
• tests (added)
• tests/Makefile.am (added)
• tests/SpreadTest.cpp (added)
• tests/test.conf (added)

trunk/thrucommon/Makefile.am

@@ -1 @@
-SUBDIRS = src
+SUBDIRS = src tests

trunk/thrucommon/configure.in

@@ -13 +13 @@
 AC_PROG_LIBTOOL
 AC_PROG_MAKE_SET
+
+# libcppunit, optional
+AC_ARG_WITH(tests,
+    AC_HELP_STRING([--with-tests],
+                   [build test suite support, requires cppunit (default is no]),
+    [with_memcached=$withval],
+    [with_memcached=no])
+if test "x$with_tests" = "xyes"; then
+    AM_PATH_CPPUNIT(1.9.6)
+    if test "x$no_cppunit" != "xyes" ; then
+        AC_DEFINE([HAVE_CPPUNIT], [1], [Define to 1 if you have the cppunit library (-lcppunit)])
+    fi
+fi
 
 AC_LANG(C++)
@@ -100 +113 @@
 AC_CONFIG_FILES([Makefile
                  src/Makefile
-                 src/thrucommon.pc])
+                 src/thrucommon.pc
+                 tests/Makefile])
 AC_OUTPUT

trunk/thrucommon/src/Makefile.am

@@ -17 +17 @@
                              ThruFileTransport.h        \
                              RecoveryManager.h          \
+                             SpreadConnection.h         \
                              SpreadManager.h            \
                              SpreadTaskFactory.h        \
@@ -35 +36 @@
                            ThruFileTransport.cpp                \
                            RecoveryManager.cpp                  \
+                           SpreadConnection.cpp                 \
                            SpreadManager.cpp                    \
                            Transaction.cpp                      \

trunk/thrucommon/src/SpreadConnection.cpp

@@ -2 +2 @@
 #include "SpreadConnection.h"
 
+using namespace log4cxx;
 using namespace std;
 
 string SP_error_to_string (int error);
+
+LoggerPtr SpreadConnection::logger (Logger::getLogger ("SpreadConnection"));
 
 SpreadConnection::SpreadConnection (const string & name,
@@ -39 +42 @@
 SpreadConnection::~SpreadConnection ()
 {
+    LOG4CXX_INFO (logger, "~SpreadConnection");
     map<string, vector<string> >::iterator i;
     for (i = this->groups.begin ();
@@ -51 +55 @@
 void SpreadConnection::join (const std::string & group)
 {
+    LOG4CXX_DEBUG (logger, "join: group=" + group);
     map<string, vector<string> >::iterator i;
     i = this->groups.find (group.c_str ());
     if (i != this->groups.end ())
     {
-        // TODO: throw an exception
+        LOG4CXX_DEBUG (logger, "    already a member");
     }
     else
@@ -70 +75 @@
         // that's handled elsewhere, as are membership changes
         this->groups[group];
+        LOG4CXX_DEBUG (logger, "    joined");
     }
 }
@@ -75 +81 @@
 void SpreadConnection::leave (const std::string & group)
 {
+    LOG4CXX_DEBUG (logger, "leave: group=" + group);
     map<string, vector<string> >::iterator i;
     i = this->groups.find (group.c_str ());
@@ -80 +87 @@
     {
         SP_leave (this->mbox, group.c_str ());
+        // TODO: this will unsubscribe everyone...
         this->groups.erase (i);
+        LOG4CXX_DEBUG (logger, "    left");
     }
 }
@@ -86 +95 @@
 void SpreadConnection::subscribe (const string & sender, const string & group,
                                   const int message_type,
-                                  subscriber_callback callback)
-{
+                                  SubscriberCallbackInfo * callback)
+{
+    char buf[256];
+    sprintf (buf, "subscribe: sender=%s, group=%s, message_type=%d", 
+             sender.c_str (), group.c_str (), message_type);
+    LOG4CXX_DEBUG (logger, buf);
     if (!group.empty ())
     {
@@ -101 +114 @@
                               const char * message, const int message_len)
 {
+    if (logger->isDebugEnabled ())
+    {
+        char buf[128];
+        sprintf (buf, "queue: group=%s, message_type=%d", group.c_str (),
+                 message_type);
+        LOG4CXX_DEBUG (logger, buf);
+    }
     QueuedMessage * queued_message =
         (QueuedMessage*)malloc (sizeof (QueuedMessage));
-    queued_message->group = group;
+    queued_message->group = strndup (group.c_str (), group.length ());
     queued_message->message_type = message_type;
     queued_message->message = strndup (message, message_len);
@@ -112 +132 @@
 void SpreadConnection::run (int count)
 {
+    if (logger->isDebugEnabled ())
+    {
+        char buf[64];
+        sprintf (buf, "run: count=%d", count);
+        LOG4CXX_DEBUG (logger, buf);
+    }
     // send out any pending message
     this->drain_pending ();
@@ -129 +155 @@
     char buf[buf_size];
 
-    int i;
+    int i = 0;
     while (!count || i < count)
     {
+        LOG4CXX_DEBUG (logger, "run:    receiving");
         buf_len = SP_receive (this->mbox, &service_type, sender,
                               max_groups, &num_groups, groups, &type,
                               &endian_mismatch, buf_size, buf);
+        // TODO: tmp
+        {
+            char buf[128];
+            sprintf (buf, "recv: buf_len=%d, sender=%s, group=%s, type=%d",
+                     buf_len, sender, groups[0], type);
+            LOG4CXX_DEBUG (logger, buf);
+        }
         if (buf_len > 0)
         {
@@ -173 +207 @@
         // drain any pending messages
         this->drain_pending ();
-    }
-}
-
-void SpreadConnection::make_callbacks (vector<subscriber_callback> callbacks,
-                                       const string & sender, 
-                                       const string & group,
-                                       const int message_type, 
-                                       const char * message,
-                                       const int message_len)
-{
+
+        i++;
+    }
+    LOG4CXX_DEBUG (logger, "run:    done");
+}
+
+void SpreadConnection::make_callbacks
+(vector<SubscriberCallbackInfo *> callbacks, const string & sender,
+ const string & group, const int message_type, const char * message,
+ const int message_len)
+{
+    if (logger->isDebugEnabled ())
+    {
+        char buf[256];
+        sprintf (buf, "make_callbacks: callbacks.size=%d, sender=%s, group=%s, message_type=%d",
+                 (int)callbacks.size (), sender.c_str (), group.c_str (), 
+                 message_type);
+        LOG4CXX_DEBUG (logger, buf);
+    }
     bool ret;
-    vector<subscriber_callback>::iterator i;
+    vector<SubscriberCallbackInfo *>::iterator i;
     for (i = callbacks.begin (); i != callbacks.end (); i++)
     {
-        ret = (*i)(this, sender, group, message_type, message, message_len);
+        subscriber_callback call = (*i)->callback;
+        void * data = (*i)->data;
+        ret = (call)(this, sender, group, message_type, message, 
+                     message_len, data);
+        ret = ((*i)->callback)(this, sender, group, message_type, message, 
+                               message_len, (*i)->data);
         if (!ret)
             callbacks.erase (i);
@@ -197 +245 @@
                                  const int message_len)
 {
+    if (logger->isDebugEnabled ())
+    {
+        char buf[128];
+        sprintf (buf, "dispatch: sender=%s, group=%s, message_type=%d",
+                 sender.c_str (), group.c_str (), message_type);
+        LOG4CXX_DEBUG (logger, buf);
+    }
+
     // if we haven't installed any callbacks, there's no reason to continue
     if (this->subscriptions.empty ())
@@ -202 +258 @@
 
     // TODO: recieve self flag
-    if (this->private_group == sender)
+    if (this->private_group == sender && !this->receive_self)
         return;
 
@@ -208 +264 @@
     // C++ is compared to perl where this is 8 lines of code :(
 
-    map<string, map<string, map<int, vector<subscriber_callback> > > >::iterator
-        s = this->subscriptions.find (sender);
+    map<string, map<string, map<int, 
+        vector<SubscriberCallbackInfo *> > > >::iterator s = 
+            this->subscriptions.find (sender);
     if (s != this->subscriptions.end ())
     {
-        map<string, map<int, vector<subscriber_callback> > >::iterator s_g =
-            (*s).second.find (group);
+        map<string, map<int, 
+            vector<SubscriberCallbackInfo *> > >::iterator s_g =
+                (*s).second.find (group);
         if (s_g != (*s).second.end ())
         {
-            map<int, vector<subscriber_callback> >::iterator s_g_t =
+            map<int, vector<SubscriberCallbackInfo *> >::iterator s_g_t =
                 (*s_g).second.find (message_type);
             if (s_g_t != (*s_g).second.end ())
             {
                 // we have a sender, group, type match
-                this->make_callbacks ((*s_g_t).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-            map<int, vector<subscriber_callback> >::iterator s_g_et = 
+                this->make_callbacks ((*s_g_t).second, sender, group,
+                                      message_type, message, message_len);
+            }
+            map<int, vector<SubscriberCallbackInfo *> >::iterator s_g_et =
                 (*s_g).second.find (-1);
             if (s_g_et != (*s_g).second.end ())
             {
                 // we have a sender, group, empty type match
-                this->make_callbacks ((*s_g_et).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-        }
-        map<string, map<int, vector<subscriber_callback> > >::iterator s_eg =
-            (*s).second.find ("");
+                this->make_callbacks ((*s_g_et).second, sender, group,
+                                      message_type, message, message_len);
+            }
+        }
+        map<string, map<int, 
+            vector<SubscriberCallbackInfo *> > >::iterator s_eg =
+                (*s).second.find ("");
         if (s_eg != (*s).second.end ())
         {
-            map<int, vector<subscriber_callback> >::iterator s_eg_t = 
+            map<int, vector<SubscriberCallbackInfo *> >::iterator s_eg_t =
                 (*s_eg).second.find (message_type);
             if (s_eg_t != (*s_eg).second.end ())
             {
                 // we have a sender, empty group, type match
-                this->make_callbacks ((*s_eg_t).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-            map<int, vector<subscriber_callback> >::iterator s_eg_et = 
+                this->make_callbacks ((*s_eg_t).second, sender, group,
+                                      message_type, message, message_len);
+            }
+            map<int, vector<SubscriberCallbackInfo *> >::iterator s_eg_et =
                 (*s_eg).second.find (-1);
             if (s_eg_et != (*s_eg).second.end ())
             {
                 // we have a sender, empty group, empty type match
-                this->make_callbacks ((*s_eg_et).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-        }
-    }
-    map<string, map<string, map<int, vector<subscriber_callback> > > >::iterator
-        es = this->subscriptions.find ("");
+                this->make_callbacks ((*s_eg_et).second, sender, group,
+                                      message_type, message, message_len);
+            }
+        }
+    }
+    map<string, map<string, map<int, 
+        vector<SubscriberCallbackInfo *> > > >::iterator
+            es = this->subscriptions.find ("");
     if (es != this->subscriptions.end ())
     {
-        map<string, map<int, vector<subscriber_callback> > >::iterator es_g =
-            (*es).second.find (group);
+        map<string, map<int, 
+            vector<SubscriberCallbackInfo *> > >::iterator es_g =
+                (*es).second.find (group);
         if (es_g != (*es).second.end ())
         {
-            map<int, vector<subscriber_callback> >::iterator es_g_t = 
+            map<int, vector<SubscriberCallbackInfo *> >::iterator es_g_t =
                 (*es_g).second.find (message_type);
             if (es_g_t != (*es_g).second.end ())
             {
                 // we have a empty sender, group, type match
-                this->make_callbacks ((*es_g_t).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-            map<int, vector<subscriber_callback> >::iterator es_g_et = 
+                this->make_callbacks ((*es_g_t).second, sender, group,
+                                      message_type, message, message_len);
+            }
+            map<int, vector<SubscriberCallbackInfo *> >::iterator es_g_et =
                 (*es_g).second.find (-1);
             if (es_g_et != (*es_g).second.end ())
             {
                 // we have a empty sender, group, empty type match
-                this->make_callbacks ((*es_g_et).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-        }
-        map<string, map<int, vector<subscriber_callback> > >::iterator es_eg =
+                this->make_callbacks ((*es_g_et).second, sender, group,
+                                      message_type, message, message_len);
+            }
+        }
+        map<string, map<int, 
+            vector<SubscriberCallbackInfo *> > >::iterator es_eg =
             (*es).second.find ("");
         if (es_eg != (*es).second.end ())
         {
-            map<int, vector<subscriber_callback> >::iterator es_eg_t = 
+            map<int, vector<SubscriberCallbackInfo *> >::iterator es_eg_t =
                 (*es_eg).second.find (message_type);
             if (es_eg_t != (*es_eg).second.end ())
             {
                 // we have a empty sender, empty group, type match
-                this->make_callbacks ((*es_eg_t).second, sender, group, 
-                                      message_type, message, message_len);
-            }
-            map<int, vector<subscriber_callback> >::iterator es_eg_et = 
+                this->make_callbacks ((*es_eg_t).second, sender, group,
+                                      message_type, message, message_len);
+            }
+            map<int, vector<SubscriberCallbackInfo *> >::iterator es_eg_et =
                 (*es_eg).second.find (-1);
             if (es_eg_et != (*es_eg).second.end ())
             {
                 // we have a empty sender, empty group, empty type match
-                this->make_callbacks ((*es_eg_et).second, sender, group, 
+                this->make_callbacks ((*es_eg_et).second, sender, group,
                                       message_type, message, message_len);
             }
@@ -306 +368 @@
 void SpreadConnection::drain_pending ()
 {
+    if (logger->isDebugEnabled ())
+    {
+        char buf[64];
+        sprintf (buf, "drain_pending: pending_messages.size=%d", 
+                 (int)this->pending_messages.size ());
+        LOG4CXX_DEBUG (logger, buf);
+    }
     int ret;
     QueuedMessage * qm;
     while (!this->pending_messages.empty ())
     {
+        LOG4CXX_DEBUG (logger, "drain_pending:    draining");
         qm = this->pending_messages.front ();
         this->pending_messages.pop ();
-        ret = SP_multicast (this->mbox, SAFE_MESS | SELF_DISCARD, 
-                            qm->group.c_str (), qm->message_type, 
+        ret = SP_multicast (this->mbox, SAFE_MESS | SELF_DISCARD,
+                            qm->group, qm->message_type,
                             qm->message_len, qm->message);
         // TODO: error handling
+        free (qm->group);
         free (qm->message);
         free (qm);
     }
+    LOG4CXX_DEBUG (logger, "drain_pending: done");
 }
 

trunk/thrucommon/src/SpreadConnection.h

@@ -17 +17 @@
 /* callbacks return true to stay installed, false to be removed */
 typedef bool (*subscriber_callback) (SpreadConnection * spread_connection,
-                                     const std::string & sender, 
-                                     const std::string & group, 
-                                     const int message_type, 
-                                     const char * message, 
-                                     const int message_len);
+                                     const std::string & sender,
+                                     const std::string & group,
+                                     const int message_type,
+                                     const char * message,
+                                     const int message_len,
+                                     void * data);
+typedef struct SubscriberCallbackInfo SubscriberCallbackInfo;
+struct SubscriberCallbackInfo
+{
+    subscriber_callback callback;
+    void * data;
+};
 
 struct QueuedMessage
 {
-    std::string group;
+    char * group;
     int message_type;
     char * message;
@@ -34 +41 @@
 {
     public:
-        SpreadConnection (const std::string & name, 
+        SpreadConnection (const std::string & name,
                           const std::string & private_name);
         ~SpreadConnection ();
 
         void subscribe (const std::string & sender, const std::string & group,
-                        const int message_type, subscriber_callback callback);
+                        const int message_type,
+                        SubscriberCallbackInfo * callback);
         void queue (const std::string & group, const int message_type,
                     const char * message, const int message_len);
         void run (int count);
+
+        void set_receive_self (bool receive_self)
+        {
+            this->receive_self = receive_self;
+        }
+
+        std::string get_private_group ()
+        {
+            return this->private_group;
+        }
 
     private:
@@ -51 +69 @@
         std::string group;
         std::string private_group;
+        bool receive_self;
         mailbox mbox;
         std::map<std::string, std::vector<std::string> > groups;
-        std::map<std::string, std::map<std::string, 
-            std::map<int, std::vector<subscriber_callback> > > > subscriptions;
+        std::map<std::string, std::map<std::string, std::map<int,
+            std::vector<SubscriberCallbackInfo *> > > > subscriptions;
         std::queue<QueuedMessage *> pending_messages;
 
@@ -62 +81 @@
                        const int message_type, const char * message,
                        const int message_len);
-        void make_callbacks (std::vector<subscriber_callback> callbacks,
-                             const std::string & sender, 
+        void make_callbacks (std::vector<SubscriberCallbackInfo *> callbacks,
+                             const std::string & sender,
                              const std::string & group,
                              const int message_type, const char * message,