minimal test for new pq event functionality

3 files changed, 83 insertions, 68 deletions

diff --git a/src/pq/pq_connect.c b/src/pq/pq_connect.c
index 00664dcd0..8722d2151 100644
--- a/src/pq/pq_connect.c
+++ b/src/pq/pq_connect.c
@@ -24,6 +24,7 @@
  */
 #include "platform.h"
 #include "pq.h"
+#include <pthread.h>
 
 
 /**
@@ -63,28 +64,6 @@ pq_notice_processor_cb (void *arg,
 }
 
 
-/**
- * Create a connection to the Postgres database using @a config_str for the
- * configuration.  Initialize logging via GNUnet's log routines and disable
- * Postgres's logger.  Also ensures that the statements in @a load_path and @a
- * es are executed whenever we (re)connect to the database, and that the
- * prepared statements in @a ps are "ready".  If statements in @es fail that
- * were created with #GNUNET_PQ_make_execute(), then the entire operation
- * fails.
- *
- * In @a load_path, a list of "$XXXX.sql" files is expected where $XXXX
- * must be a sequence of contiguous integer values starting at 0000.
- * These files are then loaded in sequence using "psql $config_str" before
- * running statements from @e es.  The directory is inspected again on
- * reconnect.
- *
- * @param config_str configuration to use
- * @param load_path path to directory with SQL transactions to run, can be NULL
- * @param es #GNUNET_PQ_PREPARED_STATEMENT_END-terminated
- *            array of statements to execute upon EACH connection, can be NULL
- * @param ps array of prepared statements to prepare, can be NULL
- * @return NULL on error
- */
 struct GNUNET_PQ_Context *
 GNUNET_PQ_connect (const char *config_str,
                    const char *load_path,
@@ -122,6 +101,11 @@ GNUNET_PQ_connect (const char *config_str,
             ps,
             plen * sizeof (struct GNUNET_PQ_PreparedStatement));
   }
+  db->channel_map = GNUNET_CONTAINER_multishortmap_create (16,
+                                                           GNUNET_YES);
+  GNUNET_assert (0 ==
+                 pthread_mutex_init (&db->notify_lock,
+                                     NULL));
   GNUNET_PQ_reconnect (db);
   if (NULL == db->conn)
   {
@@ -142,7 +126,7 @@ GNUNET_PQ_connect (const char *config_str,
  * @param i patch number to append to the @a load_path
  * @return #GNUNET_OK on success, #GNUNET_NO if patch @a i does not exist, #GNUNET_SYSERR on error
  */
-static int
+static enum GNUNET_GenericReturnValue
 apply_patch (struct GNUNET_PQ_Context *db,
              const char *load_path,
              unsigned int i)
@@ -200,15 +184,6 @@ apply_patch (struct GNUNET_PQ_Context *db,
 }
 
 
-/**
- * Within the @a db context, run all the SQL files
- * from the @a load_path from 0000-9999.sql (as long
- * as the files exist contiguously).
- *
- * @param db database context to use
- * @param load_path where to find the XXXX.sql files
- * @return #GNUNET_OK on success
- */
 enum GNUNET_GenericReturnValue
 GNUNET_PQ_run_sql (struct GNUNET_PQ_Context *db,
                    const char *load_path)
@@ -304,11 +279,6 @@ GNUNET_PQ_run_sql (struct GNUNET_PQ_Context *db,
 }
 
 
-/**
- * Reinitialize the database @a db if the connection is down.
- *
- * @param db database connection to reinitialize
- */
 void
 GNUNET_PQ_reconnect_if_down (struct GNUNET_PQ_Context *db)
 {
@@ -321,14 +291,12 @@ GNUNET_PQ_reconnect_if_down (struct GNUNET_PQ_Context *db)
 }
 
 
-/**
- * Reinitialize the database @a db.
- *
- * @param db database connection to reinitialize
- */
 void
 GNUNET_PQ_reconnect (struct GNUNET_PQ_Context *db)
 {
+  if (NULL != db->sc) 
+    db->sc (db->sc_cls,
+            -1);
   if (NULL != db->conn)
     PQfinish (db->conn);
   db->conn = PQconnectdb (db->config_str);
@@ -448,26 +416,13 @@ GNUNET_PQ_reconnect (struct GNUNET_PQ_Context *db)
     db->conn = NULL;
     return;
   }
+  if ( (NULL != db->sc) &&
+       (0 != GNUNET_CONTAINER_multishortmap_size (db->channel_map)) )
+    db->sc (db->sc_cls,
+            PQsocket (db->conn));
 }
 
 
-/**
- * Connect to a postgres database using the configuration
- * option "CONFIG" in @a section.  Also ensures that the
- * statements in @a es are executed whenever we (re)connect to the
- * database, and that the prepared statements in @a ps are "ready".
- *
- * The caller does not have to ensure that @a es and @a ps remain allocated
- * and initialized in memory until #GNUNET_PQ_disconnect() is called, as a copy will be made.
- *
- * @param cfg configuration
- * @param section configuration section to use to get Postgres configuration options
- * @param load_path_suffix suffix to append to the SQL_DIR in the configuration
- * @param es #GNUNET_PQ_PREPARED_STATEMENT_END-terminated
- *            array of statements to execute upon EACH connection, can be NULL
- * @param ps array of prepared statements to prepare, can be NULL
- * @return the postgres handle, NULL on error
- */
 struct GNUNET_PQ_Context *
 GNUNET_PQ_connect_with_cfg (const struct GNUNET_CONFIGURATION_Handle *cfg,
                             const char *section,
@@ -509,15 +464,14 @@ GNUNET_PQ_connect_with_cfg (const struct GNUNET_CONFIGURATION_Handle *cfg,
 }
 
 
-/**
- * Disconnect from the database, destroying the prepared statements
- * and releasing other associated resources.
- *
- * @param db database handle to disconnect (will be free'd)
- */
 void
 GNUNET_PQ_disconnect (struct GNUNET_PQ_Context *db)
 {
+  GNUNET_assert (0 ==
+                 GNUNET_CONTAINER_multishortmap_size (db->channel_map));
+  GNUNET_CONTAINER_multishortmap_destroy (db->channel_map);
+  GNUNET_assert (0 ==
+                 pthread_mutex_destroy (&db->notify_lock));
   GNUNET_free (db->es);
   GNUNET_free (db->ps);
   GNUNET_free (db->load_path);
diff --git a/src/pq/pq_event.c b/src/pq/pq_event.c
index ecb942230..b87aa6c5a 100644
--- a/src/pq/pq_event.c
+++ b/src/pq/pq_event.c
@@ -346,6 +346,7 @@ GNUNET_PQ_event_listen_cancel (struct GNUNET_PQ_EventHandler *eh)
   }
   GNUNET_assert (0 ==
                  pthread_mutex_unlock (&db->notify_lock));
+  GNUNET_free (eh);
 }
 
 
@@ -364,7 +365,7 @@ GNUNET_PQ_event_notify (struct GNUNET_PQ_Context *db,
   end = es_to_channel (es,
                        end);
   end = stpcpy (end,
-                "'");
+                ", '");
   end = GNUNET_STRINGS_data_to_string (extra,
                                        extra_size,
                                        end,
diff --git a/src/pq/test_pq.c b/src/pq/test_pq.c
index e588da45d..b3747dfc3 100644
--- a/src/pq/test_pq.c
+++ b/src/pq/test_pq.c
@@ -220,6 +220,64 @@ run_queries (struct GNUNET_PQ_Context *db)
 }
 
 
+static void
+event_cb (void *cls,
+          const void *extra,
+          size_t extra_size)
+{
+  unsigned int *cnt = cls;
+
+  GNUNET_assert (5 == extra_size);
+  GNUNET_assert (0 == memcmp ("world",
+                              extra,
+                              5));
+  (*cnt)++;
+}
+
+
+/**
+ * Run subscribe/notify tests.
+ *
+ * @param db database handle
+ * @return 0 on success
+ */
+static int
+test_notify (struct GNUNET_PQ_Context *db)
+{
+  struct GNUNET_PQ_EventHeaderP e1 = {
+    .size = htons (sizeof (e1)),
+    .type = htons (1)
+  };
+  struct GNUNET_PQ_EventHeaderP e2 = {
+    .size = htons (sizeof (e2)),
+    .type = htons (2)
+  };
+  unsigned int called = 0;
+  struct GNUNET_PQ_EventHandler *eh;
+
+  eh = GNUNET_PQ_event_listen (db,
+                               &e1,
+                               &event_cb,
+                               &called);
+  GNUNET_assert (NULL != eh);
+  GNUNET_PQ_event_notify (db,
+                          &e2,
+                          "hello",
+                          5);
+  GNUNET_PQ_event_do_poll (db);
+  GNUNET_assert (0 == called);
+  GNUNET_PQ_event_notify (db,
+                          &e1,
+                          "world",
+                          5);
+  GNUNET_PQ_event_do_poll (db);
+  GNUNET_assert (1 == called);
+  GNUNET_PQ_event_listen_cancel (eh);
+  return 0;
+}
+
+
+             
 int
 main (int argc,
       const char *const argv[])
@@ -272,12 +330,14 @@ main (int argc,
     return 1;
   }
   ret = run_queries (db);
+  ret |= test_notify (db);
+  ret |= test_notify (db);
 #if TEST_RESTART
   fprintf (stderr, "Please restart Postgres database now!\n");
   sleep (60);
-  ret = run_queries (db);
+  ret |= run_queries (db);
   fprintf (stderr, "Result: %d (expect: 1 -- if you restarted the DB)\n", ret);
-  ret = run_queries (db);
+  ret |= run_queries (db);
   fprintf (stderr, "Result: %d (expect: 0)\n", ret);
 #endif
   {