Introduce PART/PART_ACK messages (tests will fail until multicast module is adapted, too)

2 files changed, 111 insertions, 43 deletions

diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 73a3ae4ee..8af3264fd 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -508,8 +508,8 @@ cleanup_master (struct Master *mst)
 {
   struct Channel *chn = &mst->channel;
 
-  if (NULL != mst->origin)
-    GNUNET_MULTICAST_origin_stop (mst->origin, NULL, NULL); // FIXME
+  //if (NULL != mst->origin)
+  //  GNUNET_MULTICAST_origin_stop (mst->origin, cleanup_cb, cleanup_cls);
   GNUNET_CONTAINER_multihashmap_destroy (mst->join_reqs);
   GNUNET_CONTAINER_multihashmap_remove (masters, &chn->pub_key_hash, mst);
 }
@@ -546,11 +546,11 @@ cleanup_slave (struct Slave *slv)
     GNUNET_free (slv->relays);
     slv->relays = NULL;
   }
-  if (NULL != slv->member)
-  {
-    GNUNET_MULTICAST_member_part (slv->member, NULL, NULL); // FIXME
-    slv->member = NULL;
-  }
+  //if (NULL != slv->member)
+  //{
+  //  GNUNET_MULTICAST_member_part (slv->member, cleanup_cb, cleanup_cls);
+  //  slv->member = NULL;
+  //}
   GNUNET_CONTAINER_multihashmap_remove (slaves, &chn->pub_key_hash, slv);
 }
 
@@ -646,14 +646,15 @@ client_notify_disconnect (void *cls,
                 (GNUNET_YES == chn->is_master) ? "master" : "slave",
                 GNUNET_h2s (&chn->pub_key_hash));
     chn->is_disconnected = GNUNET_YES;
-    if (NULL != chn->tmit_head)
-    { /* Send pending messages to multicast before cleanup. */
-      transmit_message (chn);
-    }
-    else
-    {
-      cleanup_channel (chn);
-    }
+    cleanup_channel (chn);
+    //if (NULL != chn->tmit_head)
+    //{ /* Send pending messages to multicast before cleanup. */
+    //  transmit_message (chn);
+    //}
+    //else
+    //{
+    //  cleanup_channel (chn);
+    //}
   }
 }
 
@@ -2037,6 +2038,43 @@ handle_client_join_decision (void *cls,
 }
 
 
+static void
+channel_part_cb (void *cls)
+{
+  struct GNUNET_SERVICE_Client *client = cls;
+  struct GNUNET_MQ_Envelope *env;
+
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_ACK);
+  GNUNET_MQ_send (GNUNET_SERVICE_client_get_mq (client),
+                  env);
+}
+
+
+static void
+handle_client_part_request (void *cls,
+                            const struct GNUNET_MessageHeader *msg)
+{
+  struct Client *c = cls;
+
+  if (GNUNET_YES == c->channel->is_master)
+  {
+    struct Master *mst = (struct Master *) c->channel;
+
+    GNUNET_assert (NULL != mst->origin);
+    GNUNET_MULTICAST_origin_stop (mst->origin, channel_part_cb, c->client);
+    mst->origin = NULL;
+  }
+  else
+  {
+    struct Slave *slv = (struct Slave *) c->channel;
+
+    GNUNET_assert (NULL != slv->member);
+    GNUNET_MULTICAST_member_part (slv->member, channel_part_cb, c->client);
+    slv->member = NULL;
+  }
+}
+
+
 /**
  * Send acknowledgement to a client.
  *
@@ -2805,6 +2843,10 @@ GNUNET_SERVICE_MAIN
                         GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
                         struct GNUNET_PSYC_JoinDecisionMessage,
                         NULL),
+ GNUNET_MQ_hd_fixed_size (client_part_request,
+                          GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST,
+                          struct GNUNET_MessageHeader,
+                          NULL),
  GNUNET_MQ_hd_var_size (client_psyc_message,
                         GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
                         struct GNUNET_MessageHeader,
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index e7790d201..cdb9ce881 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -555,6 +555,10 @@ handle_slave_join_decision (void *cls,
 static void
 channel_cleanup (struct GNUNET_PSYC_Channel *chn)
 {
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "cleaning up channel %p\n",
+              chn);
+  GNUNET_assert (0);
   if (NULL != chn->tmit)
   {
     GNUNET_PSYC_transmit_destroy (chn->tmit);
@@ -585,33 +589,43 @@ channel_cleanup (struct GNUNET_PSYC_Channel *chn)
 
 
 static void
-channel_disconnect (struct GNUNET_PSYC_Channel *chn,
-                    GNUNET_ContinuationCallback cb,
-                    void *cls)
+handle_channel_part_ack (void *cls,
+                         const struct GNUNET_MessageHeader *msg)
 {
-  chn->is_disconnecting = GNUNET_YES;
-  chn->disconnect_cb = cb;
-  chn->disconnect_cls = cls;
+  struct GNUNET_PSYC_Channel *chn = cls;
 
-  if (NULL != chn->mq)
-  {
-    struct GNUNET_MQ_Envelope *env = GNUNET_MQ_get_last_envelope (chn->mq);
-    if (NULL != env)
-    {
-      GNUNET_MQ_notify_sent (env, (GNUNET_SCHEDULER_TaskCallback) channel_cleanup, chn);
-    }
-    else
-    {
-      channel_cleanup (chn);
-    }
-  }
-  else
-  {
-    channel_cleanup (chn);
-  }
+  channel_cleanup (chn); 
 }
 
 
+//static void
+//channel_disconnect (struct GNUNET_PSYC_Channel *chn,
+//                    GNUNET_ContinuationCallback cb,
+//                    void *cls)
+//{
+//  chn->is_disconnecting = GNUNET_YES;
+//  chn->disconnect_cb = cb;
+//  chn->disconnect_cls = cls;
+//
+//  if (NULL != chn->mq)
+//  {
+//    struct GNUNET_MQ_Envelope *env = GNUNET_MQ_get_last_envelope (chn->mq);
+//    if (NULL != env)
+//    {
+//      GNUNET_MQ_notify_sent (env, (GNUNET_SCHEDULER_TaskCallback) channel_cleanup, chn);
+//    }
+//    else
+//    {
+//      channel_cleanup (chn);
+//    }
+//  }
+//  else
+//  {
+//    channel_cleanup (chn);
+//  }
+//}
+
+
 /*** MASTER ***/
 
 
@@ -671,6 +685,10 @@ master_connect (struct GNUNET_PSYC_Master *mst)
                            GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST,
                            struct GNUNET_PSYC_JoinRequestMessage,
                            mst),
+    GNUNET_MQ_hd_fixed_size (channel_part_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
     GNUNET_MQ_hd_var_size (channel_message,
                            GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
                            struct GNUNET_PSYC_MessageHeader,
@@ -780,10 +798,12 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst,
                          void *stop_cls)
 {
   struct GNUNET_PSYC_Channel *chn = &mst->chn;
+  struct GNUNET_MQ_Envelope *env;
 
-  /* FIXME: send msg to service */
-
-  channel_disconnect (chn, stop_cb, stop_cls);
+  chn->disconnect_cb = stop_cb;
+  chn->disconnect_cls = stop_cls;
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
+  GNUNET_MQ_send (chn->mq, env);
 }
 
 
@@ -971,6 +991,10 @@ slave_connect (struct GNUNET_PSYC_Slave *slv)
                            GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION,
                            struct GNUNET_PSYC_JoinDecisionMessage,
                            slv),
+    GNUNET_MQ_hd_fixed_size (channel_part_ack,
+                             GNUNET_MESSAGE_TYPE_PSYC_PART_ACK,
+                             struct GNUNET_MessageHeader,
+                             chn),
     GNUNET_MQ_hd_var_size (channel_message,
                            GNUNET_MESSAGE_TYPE_PSYC_MESSAGE,
                            struct GNUNET_PSYC_MessageHeader,
@@ -1118,10 +1142,12 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv,
                         void *part_cls)
 {
   struct GNUNET_PSYC_Channel *chn = &slv->chn;
+  struct GNUNET_MQ_Envelope *env;
 
-  /* FIXME: send msg to service */
-
-  channel_disconnect (chn, part_cb, part_cls);
+  chn->disconnect_cb = part_cb;
+  chn->disconnect_cls = part_cls;
+  env = GNUNET_MQ_msg_header (GNUNET_MESSAGE_TYPE_PSYC_PART_REQUEST);
+  GNUNET_MQ_send (chn->mq, env);
 }