psyc service: start/stop, join/part, message transmission: lib -> psyc -> mcast; psyc API: stop/resume transmission

6 files changed, 1168 insertions, 94 deletions

diff --git a/src/psyc/Makefile.am b/src/psyc/Makefile.am
index b65f9ec7a..57f0e2280 100644
--- a/src/psyc/Makefile.am
+++ b/src/psyc/Makefile.am
@@ -24,12 +24,14 @@ libgnunetpsyc_la_SOURCES = \
   psyc.h
 libgnunetpsyc_la_LIBADD = \
   $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/env/libgnunetenv.la \
   $(GN_LIBINTL) $(XLIB)
 libgnunetpsyc_la_LDFLAGS = \
   $(GN_LIB_LDFLAGS)  $(WINFLAGS) \
   -version-info 0:0:0
 libgnunetpsyc_la_DEPENDENCIES = \
-  $(top_builddir)/src/util/libgnunetutil.la
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/env/libgnunetenv.la
 
 bin_PROGRAMS =
 
@@ -41,10 +43,14 @@ gnunet_service_psyc_SOURCES = \
 gnunet_service_psyc_LDADD = \
   $(top_builddir)/src/statistics/libgnunetstatistics.la \
   $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/multicast/libgnunetmulticast.la \
+  $(top_builddir)/src/psycstore/libgnunetpsycstore.la \
   $(GN_LIBINTL)
 gnunet_service_psyc_DEPENDENCIES = \
   $(top_builddir)/src/statistics/libgnunetstatistics.la \
-  $(top_builddir)/src/util/libgnunetutil.la
+  $(top_builddir)/src/util/libgnunetutil.la \
+  $(top_builddir)/src/multicast/libgnunetmulticast.la \
+  $(top_builddir)/src/psycstore/libgnunetpsycstore.la
 
 
 if HAVE_TESTING
diff --git a/src/psyc/gnunet-service-psyc.c b/src/psyc/gnunet-service-psyc.c
index 528b76e59..5fd2b9e66 100644
--- a/src/psyc/gnunet-service-psyc.c
+++ b/src/psyc/gnunet-service-psyc.c
@@ -29,9 +29,13 @@
 #include "gnunet_constants.h"
 #include "gnunet_protocols.h"
 #include "gnunet_statistics_service.h"
+#include "gnunet_multicast_service.h"
+#include "gnunet_psycstore_service.h"
 #include "gnunet_psyc_service.h"
 #include "psyc.h"
 
+#define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
+
 
 /**
  * Handle to our current configuration.
@@ -48,6 +52,83 @@ static struct GNUNET_STATISTICS_Handle *stats;
  */
 static struct GNUNET_SERVER_NotificationContext *nc;
 
+/**
+ * Handle to the PSYCstore.
+ */
+static struct GNUNET_PSYCSTORE_Handle *store;
+
+/**
+ * Message in the transmission queue.
+ */
+struct TransmitMessage
+{
+  struct TransmitMessage *prev;
+  struct TransmitMessage *next;
+
+  char *buf;
+  uint16_t size;
+  uint8_t status;
+};
+
+/**
+ * Common part of the client context for both a master and slave channel.
+ */
+struct Channel
+{
+  struct GNUNET_SERVER_Client *client;
+
+  struct TransmitMessage *tmit_head;
+  struct TransmitMessage *tmit_tail;
+
+  char *tmit_buf; 
+  uint32_t tmit_mod_count;
+  uint32_t tmit_mod_recvd;
+  uint16_t tmit_size;
+  uint8_t tmit_status;
+
+  uint8_t in_transmit;
+  uint8_t is_master;
+};
+
+/**
+ * Client context for a channel master.
+ */
+struct Master
+{
+  struct Channel channel;
+  struct GNUNET_CRYPTO_EccPrivateKey private_key;
+  struct GNUNET_CRYPTO_EccPublicSignKey public_key;
+
+  struct GNUNET_MULTICAST_Origin *origin;
+  struct GNUNET_MULTICAST_OriginMessageHandle *tmit_handle;
+
+  uint64_t max_message_id;
+  uint64_t max_state_message_id;
+  uint64_t max_group_generation;
+
+  /**
+   * enum GNUNET_PSYC_Policy
+   */
+  uint32_t policy;
+};
+
+
+/**
+ * Client context for a channel slave.
+ */
+struct Slave
+{
+  struct Channel channel;
+  struct GNUNET_CRYPTO_EccPrivateKey slave_key;
+  struct GNUNET_CRYPTO_EccPublicSignKey channel_key;
+
+  struct GNUNET_MULTICAST_Member *member;
+  struct GNUNET_MULTICAST_MemberRequestHandle *tmit_handle;
+
+  uint64_t max_message_id;
+  uint64_t max_request_id;
+};
+
 
 /**
  * Task run during shutdown.
@@ -70,6 +151,279 @@ shutdown_task (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   }
 }
 
+/**
+ * Called whenever a client is disconnected.
+ * Frees our resources associated with that client.
+ *
+ * @param cls Closure.
+ * @param client Identification of the client.
+ */
+static void
+client_disconnect (void *cls, struct GNUNET_SERVER_Client *client)
+{
+  if (NULL == client)
+    return;
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Client %p disconnected\n", client);
+
+  struct Channel *ch = GNUNET_SERVER_client_get_user_context (client,
+                                                              struct Channel);
+  GNUNET_assert (NULL != ch);
+
+  if (NULL != ch->tmit_buf)
+  {
+    GNUNET_free (ch->tmit_buf);
+    ch->tmit_buf = NULL;
+  }
+  GNUNET_free (ch);
+}
+
+
+void
+counters_cb (void *cls, uint64_t max_fragment_id, uint64_t max_message_id,
+             uint64_t max_group_generation, uint64_t max_state_message_id)
+{
+  struct Channel *ch = cls;
+  struct CountersResult *res = GNUNET_malloc (sizeof (*res));
+  res->header.size = htons (sizeof (*res));
+  res->max_message_id = GNUNET_htonll (max_message_id);
+
+  if (ch->is_master)
+  {
+    struct Master *mst = cls;
+    mst->max_message_id = max_message_id;
+    mst->max_state_message_id = max_state_message_id;
+    mst->max_group_generation = max_group_generation;
+    res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK);
+  }
+  else
+  {
+    struct Slave *slv = cls;
+    slv->max_message_id = max_message_id;
+    res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK);
+  }
+
+  GNUNET_SERVER_notification_context_add (nc, ch->client);
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
+                                              GNUNET_NO);
+  GNUNET_free (res);
+}
+
+
+static void
+handle_master_start (void *cls, struct GNUNET_SERVER_Client *client,
+                     const struct GNUNET_MessageHeader *msg)
+{
+  const struct MasterStartRequest *req
+    = (const struct MasterStartRequest *) msg;
+  struct Master *mst = GNUNET_new (struct Master);
+  mst->channel.client = client;
+  mst->channel.is_master = GNUNET_YES;
+  mst->policy = ntohl (req->policy);
+  mst->private_key = req->channel_key;
+  GNUNET_CRYPTO_ecc_key_get_public_for_signature (&mst->private_key,
+                                                  &mst->public_key);
+
+  GNUNET_PSYCSTORE_counters_get (store, &mst->public_key,
+                                 counters_cb, mst);
+
+  GNUNET_SERVER_client_set_user_context (client, mst);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+static void
+handle_slave_join (void *cls, struct GNUNET_SERVER_Client *client,
+                   const struct GNUNET_MessageHeader *msg)
+{
+  const struct SlaveJoinRequest *req
+    = (const struct SlaveJoinRequest *) msg;
+  struct Slave *slv = GNUNET_new (struct Slave);
+  slv->channel.client = client;
+  slv->channel.is_master = GNUNET_NO;
+  slv->channel_key = req->channel_key;
+  slv->slave_key = req->slave_key;
+
+  GNUNET_PSYCSTORE_counters_get (store, &slv->channel_key,
+                                 counters_cb, slv);
+
+  GNUNET_SERVER_client_set_user_context (client, slv);
+  GNUNET_SERVER_receive_done (client, GNUNET_OK);
+}
+
+
+static void
+send_transmit_ack (struct Channel *ch)
+{
+  struct TransmitAck *res = GNUNET_malloc (sizeof (*res));
+  res->header.size = htons (sizeof (*res));
+  res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK);
+  res->buf_avail = htons (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE - ch->tmit_size);
+
+  GNUNET_SERVER_notification_context_add (nc, ch->client);
+  GNUNET_SERVER_notification_context_unicast (nc, ch->client, &res->header,
+                                              GNUNET_NO);
+  GNUNET_free (res);
+}
+
+
+static int
+transmit_notify (void *cls, size_t *data_size, void *data)
+{
+  struct Channel *ch = cls;
+  struct TransmitMessage *msg = ch->tmit_head;
+
+  if (NULL == msg || *data_size < msg->size)
+  {
+    *data_size = 0;
+    return GNUNET_NO;
+  }
+
+  memcpy (data, msg->buf, msg->size);
+  *data_size = msg->size;
+
+  GNUNET_free (ch->tmit_buf);
+  GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, msg);
+
+  return (GNUNET_YES == ch->in_transmit) ? GNUNET_NO : GNUNET_YES;
+}
+
+
+static int
+master_transmit_message (struct Master *mst)
+{
+  if (NULL == mst->tmit_handle)
+  {
+    mst->tmit_handle
+      = GNUNET_MULTICAST_origin_to_all (mst->origin, mst->max_message_id,
+                                        mst->max_group_generation,
+                                        transmit_notify, mst);
+  }
+  else
+  {
+    GNUNET_MULTICAST_origin_to_all_resume (mst->tmit_handle);
+  }
+  return GNUNET_OK;
+}
+
+
+static int
+slave_transmit_message (struct Slave *slv)
+{
+  if (NULL == slv->tmit_handle)
+  {
+    slv->tmit_handle
+      = GNUNET_MULTICAST_member_to_origin(slv->member, slv->max_request_id,
+                                          transmit_notify, slv);
+  }
+  else
+  {
+    GNUNET_MULTICAST_member_to_origin_resume (slv->tmit_handle);
+  }
+  return GNUNET_OK;
+}
+
+
+static int
+buffer_message (struct Channel *ch, const struct GNUNET_MessageHeader *msg)
+{
+  uint16_t size = ntohs (msg->size);
+
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE < size)
+    return GNUNET_SYSERR;
+
+  if (0 == ch->tmit_size)
+  {
+    ch->tmit_buf = GNUNET_malloc (size);
+    memcpy (ch->tmit_buf, msg, size);
+    ch->tmit_size = size;
+  }
+  else if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE <= ch->tmit_size + size)
+  {
+    ch->tmit_buf = GNUNET_realloc (ch->tmit_buf, ch->tmit_size + size);
+    memcpy (ch->tmit_buf + ch->tmit_size, msg, size);
+    ch->tmit_size += size;
+  }
+
+  if (GNUNET_MULTICAST_FRAGMENT_MAX_SIZE
+      < ch->tmit_size + sizeof (struct GNUNET_PSYC_MessageData))
+  {
+    struct TransmitMessage *tmit_msg = GNUNET_new (struct TransmitMessage);
+    tmit_msg->buf = (char *) msg;
+    tmit_msg->size = size;
+    tmit_msg->status = ch->tmit_status;
+    GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, tmit_msg);
+
+    ch->is_master
+      ? master_transmit_message ((struct Master *) ch)
+      : slave_transmit_message ((struct Slave *) ch);
+  }
+
+  return GNUNET_OK;
+}
+
+static void
+handle_transmit_method (void *cls, struct GNUNET_SERVER_Client *client,
+                        const struct GNUNET_MessageHeader *msg)
+{
+  const struct GNUNET_PSYC_MessageMethod *meth
+    = (const struct GNUNET_PSYC_MessageMethod *) msg;
+  struct Channel *ch = GNUNET_SERVER_client_get_user_context (client,
+                                                              struct Channel);
+  GNUNET_assert (NULL != ch);
+
+  if (GNUNET_NO != ch->in_transmit)
+  {
+    // FIXME: already transmitting a message, send back error message.
+    return;
+  }
+
+  ch->tmit_buf = NULL;
+  ch->tmit_size = 0;
+  ch->tmit_mod_recvd = 0;
+  ch->tmit_mod_count = ntohl (meth->mod_count);
+  ch->tmit_status = GNUNET_PSYC_DATA_CONT;
+
+  buffer_message (ch, msg);
+
+  if (0 == ch->tmit_mod_count)
+    send_transmit_ack (ch);
+};
+
+
+static void
+handle_transmit_modifier (void *cls, struct GNUNET_SERVER_Client *client,
+                          const struct GNUNET_MessageHeader *msg)
+{
+  const struct GNUNET_PSYC_MessageModifier *mod
+    = (const struct GNUNET_PSYC_MessageModifier *) msg;
+  struct Channel *ch = GNUNET_SERVER_client_get_user_context (client,
+                                                              struct Channel);
+  GNUNET_assert (NULL != ch);
+
+  ch->tmit_mod_recvd++;
+  buffer_message (ch, msg);
+
+  if (ch->tmit_mod_recvd == ch->tmit_mod_count)
+    send_transmit_ack (ch);
+};
+
+
+static void
+handle_transmit_data (void *cls, struct GNUNET_SERVER_Client *client,
+                      const struct GNUNET_MessageHeader *msg)
+{
+  const struct GNUNET_PSYC_MessageData *data
+    = (const struct GNUNET_PSYC_MessageData *) msg;
+  struct Channel *ch = GNUNET_SERVER_client_get_user_context (client,
+                                                              struct Channel);
+  GNUNET_assert (NULL != ch);
+
+  ch->tmit_status = data->status;
+  buffer_message (ch, msg);
+  send_transmit_ack (ch);
+};
+
 
 /**
  * Initialize the PSYC service.
@@ -83,14 +437,30 @@ run (void *cls, struct GNUNET_SERVER_Handle *server,
      const struct GNUNET_CONFIGURATION_Handle *c)
 {
   static const struct GNUNET_SERVER_MessageHandler handlers[] = {
+    { &handle_master_start, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_MASTER_START, 0 },
+
+    { &handle_slave_join, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN, 0 },
+
+    { &handle_transmit_method, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_METHOD, 0 },
+
+    { &handle_transmit_modifier, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_MODIFIER, 0 },
+
+    { &handle_transmit_data, NULL,
+      GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_DATA, 0 },
+
     { NULL, NULL, 0, 0 }
   };
 
   cfg = c;
-
+  store = GNUNET_PSYCSTORE_connect (cfg);
   stats = GNUNET_STATISTICS_create ("psyc", cfg);
-  GNUNET_SERVER_add_handlers (server, handlers);
   nc = GNUNET_SERVER_notification_context_create (server, 1);
+  GNUNET_SERVER_add_handlers (server, handlers);
+  GNUNET_SERVER_disconnect_notify (server, &client_disconnect, NULL);
   GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_UNIT_FOREVER_REL, &shutdown_task,
                                 NULL);
 }
diff --git a/src/psyc/psyc.conf b/src/psyc/psyc.conf
new file mode 100644
index 000000000..5a1eebf61
--- /dev/null
+++ b/src/psyc/psyc.conf
@@ -0,0 +1,7 @@
+[psyc]
+AUTOSTART = YES
+HOME = $SERVICEHOME
+BINARY = gnunet-service-psyc
+UNIXPATH = /tmp/gnunet-service-psyc.sock
+UNIX_MATCH_UID = NO
+UNIX_MATCH_GID = YES
diff --git a/src/psyc/psyc.h b/src/psyc/psyc.h
index c82c10ffa..35e9ae800 100644
--- a/src/psyc/psyc.h
+++ b/src/psyc/psyc.h
@@ -24,13 +24,186 @@
  * @author Gabor X Toth
  */
 
-#ifndef GNUNET_PSYC_H
-#define GNUNET_PSYC_H
+#ifndef PSYC_H
+#define PSYC_H
 
 #include "gnunet_common.h"
 
 GNUNET_NETWORK_STRUCT_BEGIN
 
+/**** service -> library ****/
+
+/**
+ * Answer from service to client about last operation.
+ */
+struct OperationResult
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYCSTORE_RESULT_CODE
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Operation ID.
+   */
+  uint32_t op_id GNUNET_PACKED;
+
+  /**
+   * Status code for the operation.
+   */
+  int64_t result_code GNUNET_PACKED;
+
+  /* followed by 0-terminated error message (on error) */
+
+};
+
+
+struct CountersResult
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_RESULT_COUNTERS
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint64_t max_message_id;
+};
+
+
+/**
+ * Transmit acknowledgment.
+ *
+ * Sent after the last GNUNET_PSYC_MessageModifier and after each
+ * GNUNET_PSYC_MessageData.
+ *
+ * This message acknowledges previously received messages and asks for the next
+ * fragment of data.
+ */
+struct TransmitAck
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Buffer space available for the next data fragment.
+   */
+  uint16_t buf_avail;
+};
+
+
+/**** library -> service ****/
+
+
+struct MasterStartRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_MASTER_START
+   */
+  struct GNUNET_MessageHeader header;
+
+  struct GNUNET_CRYPTO_EccPrivateKey channel_key;
+
+  uint32_t policy GNUNET_PACKED;
+};
+
+
+struct SlaveJoinRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t relay_count GNUNET_PACKED;
+
+  struct GNUNET_CRYPTO_EccPublicSignKey channel_key;
+
+  struct GNUNET_CRYPTO_EccPrivateKey slave_key;
+
+  /* Followed by struct GNUNET_PeerIdentity relays[relay_count] */
+};
+
+
+struct ChannelSlaveAdd
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t reserved;
+
+  struct GNUNET_CRYPTO_EccPublicSignKey *slave_key;
+
+  uint64_t announced_at;
+
+  uint64_t effective_since;
+};
+
+
+struct ChannelSlaveRemove
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint32_t reserved;
+
+  struct GNUNET_CRYPTO_EccPublicSignKey *slave_key;
+
+  uint64_t announced_at;
+};
+
+
+struct StoryRequest
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STORY_REQUEST
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint64_t op_id;
+
+  uint64_t start_message_id;
+
+  uint64_t end_message_id;
+};
+
+
+struct StateQuery
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STATE_QUERY
+   */
+  struct GNUNET_MessageHeader header;
+
+  uint64_t op_id;
+
+  /* Followed by NUL-terminated name. */
+};
+
+
+struct StateResult
+{
+  /**
+   * Type: GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_STATE_RESULT
+   */
+  struct GNUNET_MessageHeader header;
+
+  /**
+   * Size of name, including NUL terminator.
+   */
+  uint16_t name_size GNUNET_PACKED;
+
+  /**
+   * OR'd StateOpFlags
+   */
+  uint8_t flags;
+
+  /* Followed by NUL-terminated name, then the value. */
+};
 
 
 GNUNET_NETWORK_STRUCT_END
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c
index e298fae71..61b57c050 100644
--- a/src/psyc/psyc_api.c
+++ b/src/psyc/psyc_api.c
@@ -36,15 +36,76 @@
 #include "gnunet_psyc_service.h"
 #include "psyc.h"
 
+#define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__)
+
+
+struct OperationHandle
+{
+  struct OperationHandle *prev;
+  struct OperationHandle *next;
+  const struct GNUNET_MessageHeader *msg;
+};
+
 /** 
- * Handle that identifies a join request.
- *
- * Used to match calls to #GNUNET_PSYC_JoinCallback to the
- * corresponding calls to GNUNET_PSYC_join_decision().
+ * Handle to access PSYC channel operations for both the master and slaves.
  */
-struct GNUNET_PSYC_JoinHandle
+struct GNUNET_PSYC_Channel
 {
-
+  /**
+   * Configuration to use.
+   */
+  const struct GNUNET_CONFIGURATION_Handle *cfg;
+
+  /**
+   * Socket (if available).
+   */
+  struct GNUNET_CLIENT_Connection *client;
+
+  /**
+   * Currently pending transmission request, or NULL for none.
+   */
+  struct GNUNET_CLIENT_TransmitHandle *th;
+
+  /**
+   * Head of operations to transmit.
+   */
+  struct OperationHandle *transmit_head;
+
+  /**
+   * Tail of operations to transmit.
+   */
+  struct OperationHandle *transmit_tail;
+
+  /**
+   * Message to send on reconnect.
+   */
+  struct GNUNET_MessageHeader *reconnect_msg;
+
+  /**
+   * Task doing exponential back-off trying to reconnect.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
+
+  /**
+   * Time for next connect retry.
+   */
+  struct GNUNET_TIME_Relative reconnect_delay;
+
+  GNUNET_PSYC_Method method_cb;
+
+  GNUNET_PSYC_JoinCallback join_cb;
+
+  void *cb_cls;
+
+  /**
+   * Are we polling for incoming messages right now?
+   */
+  int in_receive;
+
+  /**
+   * Are we currently transmitting a message?
+   */
+  int in_transmit;
 };
 
 
@@ -53,23 +114,30 @@ struct GNUNET_PSYC_JoinHandle
  */
 struct GNUNET_PSYC_Master
 {
+  struct GNUNET_PSYC_Channel ch;
+
+  GNUNET_PSYC_MasterStartCallback start_cb;
 
+  uint64_t max_message_id;
 };
 
 
 /** 
- * Handle for a pending PSYC transmission operation.
+ * Handle for a PSYC channel slave.
  */
-struct GNUNET_PSYC_MasterTransmitHandle
+struct GNUNET_PSYC_Slave
 {
-
+  struct GNUNET_PSYC_Channel ch;
 };
 
 
 /** 
- * Handle for a PSYC channel slave.
+ * Handle that identifies a join request.
+ *
+ * Used to match calls to #GNUNET_PSYC_JoinCallback to the
+ * corresponding calls to GNUNET_PSYC_join_decision().
  */
-struct GNUNET_PSYC_Slave
+struct GNUNET_PSYC_JoinHandle
 {
 
 };
@@ -78,16 +146,20 @@ struct GNUNET_PSYC_Slave
 /** 
  * Handle for a pending PSYC transmission operation.
  */
-struct GNUNET_PSYC_SlaveTransmitHandle
+struct GNUNET_PSYC_MasterTransmitHandle
 {
-
+  struct GNUNET_PSYC_Master *master;
+  const struct GNUNET_ENV_Environment *env;
+  GNUNET_PSYC_MasterTransmitNotify notify;
+  void *notify_cls;
+  enum GNUNET_PSYC_MasterTransmitFlags flags;
 };
 
 
 /** 
- * Handle to access PSYC channel operations for both the master and slaves.
+ * Handle for a pending PSYC transmission operation.
  */
-struct GNUNET_PSYC_Channel
+struct GNUNET_PSYC_SlaveTransmitHandle
 {
 
 };
@@ -102,45 +174,264 @@ struct GNUNET_PSYC_Story
 };
 
 
+/**
+ * Handle for a state query operation.
+ */
 struct GNUNET_PSYC_StateQuery
 {
 
 };
 
 
-/** 
- * Function to call with the decision made for a join request.
+/**
+ * Try again to connect to the PSYCstore service.
  *
- * Must be called once and only once in response to an invocation of the
- * #GNUNET_PSYC_JoinCallback.
+ * @param cls handle to the PSYCstore service.
+ * @param tc scheduler context
+ */
+static void
+reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
+
+
+/**
+ * Reschedule a connect attempt to the service.
  *
- * @param jh Join request handle.
- * @param is_admitted #GNUNET_YES if joining is approved,
- *        #GNUNET_NO if it is disapproved.
- * @param relay_count Number of relays given.
- * @param relays Array of suggested peers that might be useful relays to use
- *        when joining the multicast group (essentially a list of peers that
- *        are already part of the multicast group and might thus be willing
- *        to help with routing).  If empty, only this local peer (which must
- *        be the multicast origin) is a good candidate for building the
- *        multicast tree.  Note that it is unnecessary to specify our own
- *        peer identity in this array.
- * @param method_name Method name for the message transmitted with the response.
- * @param env Environment containing transient variables for the message, or NULL.
- * @param data Data of the message.
- * @param data_size Size of @a data.
+ * @param h transport service to reconnect
  */
-void
-GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
-                           int is_admitted,
-                           unsigned int relay_count,
-                           const struct GNUNET_PeerIdentity *relays,
-                           const char *method_name,
-                           const struct GNUNET_ENV_Environment *env,
-                           const void *data,
-                           size_t data_size)
+static void
+reschedule_connect (struct GNUNET_PSYC_Channel *c)
 {
+  GNUNET_assert (c->reconnect_task == GNUNET_SCHEDULER_NO_TASK);
+
+  if (NULL != c->th)
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (c->th);
+    c->th = NULL;
+  }
+  if (NULL != c->client)
+  {
+    GNUNET_CLIENT_disconnect (c->client);
+    c->client = NULL;
+  }
+  c->in_receive = GNUNET_NO;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Scheduling task to reconnect to PSYCstore service in %s.\n",
+       GNUNET_STRINGS_relative_time_to_string (c->reconnect_delay, GNUNET_YES));
+  c->reconnect_task =
+      GNUNET_SCHEDULER_add_delayed (c->reconnect_delay, &reconnect, c);
+  c->reconnect_delay = GNUNET_TIME_STD_BACKOFF (c->reconnect_delay);
+}
+
+
+/**
+ * Schedule transmission of the next message from our queue.
+ *
+ * @param h PSYCstore handle
+ */
+static void
+transmit_next (struct GNUNET_PSYC_Channel *c);
+
+
+/**
+ * Type of a function to call when we receive a message
+ * from the service.
+ *
+ * @param cls closure
+ * @param msg message received, NULL on timeout or fatal error
+ */
+static void
+message_handler (void *cls, const struct GNUNET_MessageHeader *msg)
+{
+  struct GNUNET_PSYC_Channel *ch = cls;
+  struct GNUNET_PSYC_Master *mst = cls;
+  struct GNUNET_PSYC_Slave *slv = cls;
+
+  if (NULL == msg)
+  {
+    reschedule_connect (ch);
+    return;
+  }
+  uint16_t size_eq = 0;
+  uint16_t size_min = 0;
+  const uint16_t size = ntohs (msg->size);
+  const uint16_t type = ntohs (msg->type);
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Received message of type %d from PSYC service\n", type);
+
+  switch (type)
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK:
+  case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
+    size_eq = sizeof (struct CountersResult);
+    break;
+  }
+
+  if (! ((0 < size_eq && size == size_eq)
+         || (0 < size_min && size >= size_min)))
+  {
+    GNUNET_break (0);
+    reschedule_connect (ch);
+    return;
+  }
+
+  struct CountersResult *cres;
+
+  switch (type)
+  {
+  case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK:
+    cres = (struct CountersResult *) msg;
+    mst->max_message_id = GNUNET_ntohll (cres->max_message_id);
+    if (NULL != mst->start_cb)
+      mst->start_cb (ch->cb_cls, mst->max_message_id);
+    break;
+
+  case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK:
+    cres = (struct CountersResult *) msg;
+#if TODO
+    slv->max_message_id = GNUNET_ntohll (cres->max_message_id);
+    if (NULL != slv->join_ack_cb)
+      mst->join_ack_cb (ch->cb_cls, mst->max_message_id);
+#endif
+    break;
+  }
+
+  GNUNET_CLIENT_receive (ch->client, &message_handler, ch,
+                         GNUNET_TIME_UNIT_FOREVER_REL);
+}
 
+
+/**
+ * Transmit next message to service.
+ *
+ * @param cls The 'struct GNUNET_PSYCSTORE_Handle'.
+ * @param size Number of bytes available in buf.
+ * @param buf Where to copy the message.
+ * @return Number of bytes copied to buf.
+ */
+static size_t
+send_next_message (void *cls, size_t size, void *buf)
+{
+  struct GNUNET_PSYC_Channel *ch = cls;
+  struct OperationHandle *op = ch->transmit_head;
+  size_t ret;
+
+  ch->th = NULL;
+  if (NULL == op->msg)
+    return 0;
+  ret = ntohs (op->msg->size);
+  if (ret > size)
+  {
+    reschedule_connect (ch);
+    return 0;
+  }
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Sending message of type %d to PSYCstore service\n",
+       ntohs (op->msg->type));
+  memcpy (buf, op->msg, ret);
+
+  GNUNET_CONTAINER_DLL_remove (ch->transmit_head, ch->transmit_tail, op);
+  GNUNET_free (op);
+
+  if (NULL != ch->transmit_head)
+    transmit_next (ch);
+
+  if (GNUNET_NO == ch->in_receive)
+  {
+    ch->in_receive = GNUNET_YES;
+    GNUNET_CLIENT_receive (ch->client, &message_handler, ch,
+                           GNUNET_TIME_UNIT_FOREVER_REL);
+  }
+  return ret;
+}
+
+
+/**
+ * Schedule transmission of the next message from our queue.
+ *
+ * @param h PSYCstore handle.
+ */
+static void
+transmit_next (struct GNUNET_PSYC_Channel *ch)
+{
+  if (NULL != ch->th || NULL == ch->client)
+    return;
+
+  struct OperationHandle *op = ch->transmit_head;
+  if (NULL == op)
+    return;
+
+  ch->th = GNUNET_CLIENT_notify_transmit_ready (ch->client,
+                                                ntohs (op->msg->size),
+                                                GNUNET_TIME_UNIT_FOREVER_REL,
+                                                GNUNET_NO,
+                                                &send_next_message,
+                                                ch);
+}
+
+
+/**
+ * Try again to connect to the PSYC service.
+ *
+ * @param cls Channel handle.
+ * @param tc Scheduler context.
+ */
+static void
+reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_PSYC_Channel *ch = cls;
+
+  ch->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Connecting to PSYC service.\n");
+  GNUNET_assert (NULL == ch->client);
+  ch->client = GNUNET_CLIENT_connect ("psyc", ch->cfg);
+  GNUNET_assert (NULL != ch->client);
+
+  if (NULL == ch->transmit_head ||
+      ch->transmit_head->msg->type != ch->reconnect_msg->type)
+  {
+    struct OperationHandle *op
+      = GNUNET_malloc (sizeof (struct OperationHandle)
+                       + ntohs (ch->reconnect_msg->size));
+    memcpy (&op[1], ch->reconnect_msg, ntohs (ch->reconnect_msg->size));
+    op->msg = (struct GNUNET_MessageHeader *) &op[1];
+    GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+  }
+
+  transmit_next (ch);
+}
+
+
+/**
+ * Disconnect from the PSYC service.
+ *
+ * @param cls Channel handle.
+ * @param tc Scheduler context.
+ */
+static void
+disconnect (void *c)
+{
+  struct GNUNET_PSYC_Channel *ch = c;
+  GNUNET_assert (NULL != ch);
+  GNUNET_assert (ch->transmit_head == ch->transmit_tail);
+  if (ch->reconnect_task != GNUNET_SCHEDULER_NO_TASK)
+  {
+    GNUNET_SCHEDULER_cancel (ch->reconnect_task);
+    ch->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
+  }
+  if (NULL != ch->th)
+  {
+    GNUNET_CLIENT_notify_transmit_ready_cancel (ch->th);
+    ch->th = NULL;
+  }
+  if (NULL != ch->client)
+  {
+    GNUNET_CLIENT_disconnect (ch->client);
+    ch->client = NULL;
+  }
+  if (NULL != ch->reconnect_msg)
+    ch->reconnect_msg = NULL;
 }
 
 
@@ -177,57 +468,172 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg,
                           enum GNUNET_PSYC_Policy policy,
                           GNUNET_PSYC_Method method,
                           GNUNET_PSYC_JoinCallback join_cb,
+                          GNUNET_PSYC_MasterStartCallback master_started_cb,
                           void *cls)
 {
-  return NULL;
+  struct GNUNET_PSYC_Master *mst = GNUNET_malloc (sizeof (*mst));
+  struct GNUNET_PSYC_Channel *ch = &mst->ch;
+  struct MasterStartRequest *req = GNUNET_malloc (sizeof (*req));
+
+  req->header.size = htons (sizeof (*req) + sizeof (*channel_key));
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START);
+  req->channel_key = *channel_key;
+  req->policy = policy;
+
+  ch->cfg = cfg;
+  ch->reconnect_msg = (struct GNUNET_MessageHeader *) req;
+  ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
+  ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, mst);
+
+  ch->method_cb = method;
+  ch->join_cb = join_cb;
+  ch->cb_cls = cls;
+  mst->start_cb = master_started_cb;
+
+  return mst;
+}
+
+
+/** 
+ * Stop a PSYC master channel.
+ *
+ * @param master PSYC channel master to stop.
+ */
+void
+GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst)
+{
+  disconnect (mst);
+  GNUNET_free (mst);
+}
+
+
+/** 
+ * Function to call with the decision made for a join request.
+ *
+ * Must be called once and only once in response to an invocation of the
+ * #GNUNET_PSYC_JoinCallback.
+ *
+ * @param jh Join request handle.
+ * @param is_admitted #GNUNET_YES if joining is approved,
+ *        #GNUNET_NO if it is disapproved.
+ * @param relay_count Number of relays given.
+ * @param relays Array of suggested peers that might be useful relays to use
+ *        when joining the multicast group (essentially a list of peers that
+ *        are already part of the multicast group and might thus be willing
+ *        to help with routing).  If empty, only this local peer (which must
+ *        be the multicast origin) is a good candidate for building the
+ *        multicast tree.  Note that it is unnecessary to specify our own
+ *        peer identity in this array.
+ * @param method_name Method name for the message transmitted with the response.
+ * @param env Environment containing transient variables for the message, or NULL.
+ * @param data Data of the message.
+ * @param data_size Size of @a data.
+ */
+void
+GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh,
+                           int is_admitted,
+                           unsigned int relay_count,
+                           const struct GNUNET_PeerIdentity *relays,
+                           const char *method_name,
+                           const struct GNUNET_ENV_Environment *env,
+                           const void *data,
+                           size_t data_size)
+{
+
+}
+
+
+/* FIXME: split up value into <64K chunks and transmit the continuations in
+ *        MOD_CONT msgs */
+int
+send_modifier (void *cls, struct GNUNET_ENV_Modifier *mod)
+{
+  struct GNUNET_PSYC_Channel *ch = cls;
+  size_t name_size = strlen (mod->name) + 1;
+  struct GNUNET_PSYC_MessageModifier *pmod;
+  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*pmod)
+                                              + name_size + mod->value_size);
+  pmod = (struct GNUNET_PSYC_MessageModifier *) &op[1];
+  op->msg = (struct GNUNET_MessageHeader *) pmod;
+
+  pmod->header.type = GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_MODIFIER;
+  pmod->header.size = htons (sizeof (*pmod) + name_size + mod->value_size);
+  pmod->name_size = htons (name_size);
+  memcpy (&pmod[1], mod->name, name_size);
+  memcpy ((void *) &pmod[1] + name_size, mod->value, mod->value_size);
+
+  GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+  return GNUNET_YES;
 }
 
 
 /** 
  * Send a message to call a method to all members in the PSYC channel.
  *
- * @param master Handle to the PSYC channel.
+ * @param mst Handle to the PSYC channel.
  * @param method_name Which method should be invoked.
  * @param env Environment containing state operations and transient variables
  *            for the message, or NULL.
  * @param notify Function to call to obtain the arguments.
  * @param notify_cls Closure for @a notify.
  * @param flags Flags for the message being transmitted.
- * @return Transmission handle, NULL on error (i.e. more than one request queued).
+ * @return Transmission handle, NULL on error (i.e. more than one request
+ *         queued).
  */
 struct GNUNET_PSYC_MasterTransmitHandle *
-GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master,
+GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *mst,
                              const char *method_name,
                              const struct GNUNET_ENV_Environment *env,
                              GNUNET_PSYC_MasterTransmitNotify notify,
                              void *notify_cls,
                              enum GNUNET_PSYC_MasterTransmitFlags flags)
 {
-  return NULL;
+  GNUNET_assert (NULL != mst);
+  struct GNUNET_PSYC_Channel *ch = &mst->ch;
+  if (GNUNET_NO != ch->in_transmit)
+    return NULL;
+  ch->in_transmit = GNUNET_YES;
+
+  struct GNUNET_PSYC_MessageMethod *pmeth;
+  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*pmeth));
+  pmeth = (struct GNUNET_PSYC_MessageMethod *) &op[1];
+  op->msg = (struct GNUNET_MessageHeader *) pmeth;
+
+  pmeth->header.type = GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_METHOD;
+  size_t size = strlen (method_name) + 1;
+  pmeth->header.size = htons (sizeof (*pmeth) + size);
+  pmeth->flags = htonl (flags);
+  pmeth->mod_count
+    = GNUNET_ntohll (GNUNET_ENV_environment_get_mod_count (env));
+  memcpy (&pmeth[1], method_name, size);
+
+  GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+
+  GNUNET_ENV_environment_iterate (env, send_modifier, mst);
+  
+  struct GNUNET_PSYC_MasterTransmitHandle *th = GNUNET_malloc (sizeof (*th));
+  th->master = mst;
+  th->env = env;
+  th->notify = notify;
+  th->notify_cls = notify_cls;
+  return th;
 }
 
 
 /** 
- * Abort transmission request to channel.
+ * Abort transmission request to the channel.
  *
  * @param th Handle of the request that is being aborted.
  */
 void
 GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th)
 {
+  struct GNUNET_PSYC_Master *mst = th->master;
+  struct GNUNET_PSYC_Channel *ch = &mst->ch;
+  if (GNUNET_NO != ch->in_transmit)
+    return;
 
-}
-
-
-/** 
- * Stop a PSYC master channel.
- *
- * @param master PSYC channel master to stop.
- */
-void
-GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master)
-{
-
+  
 }
 
 
@@ -235,7 +641,7 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master)
  * Join a PSYC channel.
  *
  * The entity joining is always the local peer.  The user must immediately use
- * the GNUNET_PSYC_slave_to_master() functions to transmit a @e join_msg to the
+ * the GNUNET_PSYC_slave_transmit() functions to transmit a @e join_msg to the
  * channel; if the join request succeeds, the channel state (and @e recent
  * method calls) will be replayed to the joining member.  There is no explicit
  * notification on failure (as the channel may simply take days to approve,
@@ -269,13 +675,32 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
                         const struct GNUNET_PeerIdentity *relays,
                         GNUNET_PSYC_Method method,
                         GNUNET_PSYC_JoinCallback join_cb,
+                        GNUNET_PSYC_SlaveJoinCallback slave_joined_cb,
                         void *cls,
                         const char *method_name,
                         const struct GNUNET_ENV_Environment *env,
                         const void *data,
                         size_t data_size)
 {
-  return NULL;
+  struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv));
+  struct GNUNET_PSYC_Channel *ch = &slv->ch;
+  struct SlaveJoinRequest *req = GNUNET_malloc (sizeof (*req));
+
+  req->header.size = htons (sizeof (*req)
+                            + sizeof (*channel_key) + sizeof (*slave_key)
+                            + relay_count * sizeof (*relays));
+  req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN);
+  req->channel_key = *channel_key;
+  req->slave_key = *slave_key;
+  req->relay_count = relay_count;
+  memcpy (&req[1], relays, relay_count * sizeof (*relays));
+
+  ch->cfg = cfg;
+  ch->reconnect_msg = (struct GNUNET_MessageHeader *) req;
+  ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO;
+  ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, slv);
+
+  return slv;
 }
 
 
@@ -283,14 +708,15 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg,
  * Part a PSYC channel.
  *
  * Will terminate the connection to the PSYC service.  Polite clients should
- * first explicitly send a @e part request (via GNUNET_PSYC_slave_to_master()).
+ * first explicitly send a @e part request (via GNUNET_PSYC_slave_transmit()).
  *
- * @param slave Slave handle.
+ * @param slv Slave handle.
  */
 void
-GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave)
+GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv)
 {
-
+  disconnect (slv);
+  GNUNET_free (slv);
 }
 
 
@@ -299,11 +725,13 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave)
  *
  * @param slave Slave handle.
  * @param method_name Which (PSYC) method should be invoked (on host).
- * @param env Environment containing transient variables for the message, or NULL.
+ * @param env Environment containing transient variables for the message, or
+ *            NULL.
  * @param notify Function to call when we are allowed to transmit (to get data).
  * @param notify_cls Closure for @a notify.
  * @param flags Flags for the message being transmitted.
- * @return Transmission handle, NULL on error (i.e. more than one request queued).
+ * @return Transmission handle, NULL on error (i.e. more than one request
+ *         queued).
  */
 struct GNUNET_PSYC_SlaveTransmitHandle *
 GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave,
@@ -330,7 +758,8 @@ GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th)
 
 
 /** 
- * Convert a channel @a master to a @e channel handle to access the @e channel APIs.
+ * Convert a channel @a master to a @e channel handle to access the @e channel
+ * APIs.
  *
  * @param master Channel master handle.
  * @return Channel handle, valid for as long as @a master is valid.
@@ -338,7 +767,7 @@ GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th)
 struct GNUNET_PSYC_Channel *
 GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
 {
-  return NULL;
+  return (struct GNUNET_PSYC_Channel *) master;
 }
 
 
@@ -351,7 +780,7 @@ GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master)
 struct GNUNET_PSYC_Channel *
 GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slave)
 {
-  return NULL;
+  return (struct GNUNET_PSYC_Channel *) slave;
 }
 
 
@@ -371,18 +800,30 @@ GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slave)
  * correctly; not doing so correctly will result in either denying other slaves
  * access or offering access to channel data to non-members.
  *
- * @param channel Channel handle.
+ * @param ch Channel handle.
  * @param slave_key Identity of channel slave to add.
  * @param announced_at ID of the message that announced the membership change.
  * @param effective_since Addition of slave is in effect since this message ID.
  */
 void
-GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel,
-                               const struct GNUNET_CRYPTO_EccPublicSignKey *slave_key,
+GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *ch,
+                               const struct GNUNET_CRYPTO_EccPublicSignKey
+                               *slave_key,
                                uint64_t announced_at,
                                uint64_t effective_since)
 {
-
+  struct ChannelSlaveAdd *slvadd;
+  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvadd));
+  slvadd = (struct ChannelSlaveAdd *) &op[1];
+  op->msg = (struct GNUNET_MessageHeader *) slvadd;
+
+  slvadd->header.type = GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD;
+  slvadd->header.size = htons (sizeof (*slvadd));
+  slvadd->announced_at = GNUNET_htonll (announced_at);
+  slvadd->effective_since = GNUNET_htonll (effective_since);
+
+  GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+  transmit_next (ch);
 }
 
 
@@ -403,16 +844,27 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel,
  * denying members access or offering access to channel data to
  * non-members.
  *
- * @param channel Channel handle.
+ * @param ch Channel handle.
  * @param slave_key Identity of channel slave to remove.
  * @param announced_at ID of the message that announced the membership change.
  */
 void
-GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
-                                  const struct GNUNET_CRYPTO_EccPublicSignKey *slave_key,
+GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *ch,
+                                  const struct GNUNET_CRYPTO_EccPublicSignKey
+                                  *slave_key,
                                   uint64_t announced_at)
 {
+  struct ChannelSlaveRemove *slvrm;
+  struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvrm));
+  slvrm = (struct ChannelSlaveRemove *) &op[1];
+  op->msg = (struct GNUNET_MessageHeader *) slvrm;
+
+  slvrm->header.type = GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM;
+  slvrm->header.size = htons (sizeof (*slvrm));
+  slvrm->announced_at = GNUNET_htonll (announced_at);
 
+  GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op);
+  transmit_next (ch);
 }
 
 
@@ -424,7 +876,7 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
  *
  * To get the latest message, use 0 for both the start and end message ID.
  *
- * @param channel Which channel should be replayed?
+ * @param ch Which channel should be replayed?
  * @param start_message_id Earliest interesting point in history.
  * @param end_message_id Last (exclusive) interesting point in history.
  * @param method Function to invoke on messages received from the story.
@@ -441,7 +893,7 @@ GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel,
  * @return Handle to cancel story telling operation.
  */
 struct GNUNET_PSYC_Story *
-GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel,
+GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *ch,
                                 uint64_t start_message_id,
                                 uint64_t end_message_id,
                                 GNUNET_PSYC_Method method,
@@ -495,9 +947,9 @@ GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel,
 /** 
  * Return all channel state variables whose name matches a given prefix.
  *
- * A name matches if it starts with the given @a name_prefix, thus requesting the
- * empty prefix ("") will match all values; requesting "_a_b" will also return
- * values stored under "_a_b_c".
+ * A name matches if it starts with the given @a name_prefix, thus requesting
+ * the empty prefix ("") will match all values; requesting "_a_b" will also
+ * return values stored under "_a_b_c".
  *
  * The @a state_cb is invoked on all matching state variables asynchronously, as
  * the state is stored in and retrieved from the PSYCstore,
diff --git a/src/psyc/test_psyc.c b/src/psyc/test_psyc.c
index b37b3ceb1..84c9840f5 100644
--- a/src/psyc/test_psyc.c
+++ b/src/psyc/test_psyc.c
@@ -28,12 +28,12 @@
 #include "platform.h"
 #include "gnunet_common.h"
 #include "gnunet_util_lib.h"
-#include "gnunet_psycstore_service.h"
 #include "gnunet_testing_lib.h"
+#include "gnunet_psyc_service.h"
 
 #define TIMEOUT GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 10)
 
-#define DEBUG_SERVICE 0
+#define DEBUG_SERVICE 1
 
 
 /**
@@ -41,11 +41,22 @@
  */
 static int res;
 
+static const struct GNUNET_CONFIGURATION_Handle *cfg;
+
 /**
  * Handle for task for timeout termination.
  */
 static GNUNET_SCHEDULER_TaskIdentifier end_badly_task;
 
+static struct GNUNET_PSYC_Master *mst;
+static struct GNUNET_PSYC_Slave *slv;
+static struct GNUNET_PSYC_Channel *ch;
+
+static struct GNUNET_CRYPTO_EccPrivateKey *channel_key;
+static struct GNUNET_CRYPTO_EccPrivateKey *slave_key;
+
+static struct GNUNET_CRYPTO_EccPublicSignKey channel_pub_key;
+static struct GNUNET_CRYPTO_EccPublicSignKey slave_pub_key;
 
 /**
  * Clean up all resources used.
@@ -53,6 +64,11 @@ static GNUNET_SCHEDULER_TaskIdentifier end_badly_task;
 static void
 cleanup ()
 {
+  if (master != NULL)
+  {
+    GNUNET_PSYC_master_stop (master);
+    master = NULL;
+  }
   GNUNET_SCHEDULER_shutdown ();
 }
 
@@ -100,6 +116,42 @@ end ()
                                 &end_normally, NULL);
 }
 
+
+static int
+method (void *cls, const struct GNUNET_CRYPTO_EccPublicSignKey *slave_key,
+        uint64_t message_id, const char *method_name,
+        size_t modifier_count, const struct GNUNET_ENV_Modifier *modifiers,
+        uint64_t data_offset, const void *data, size_t data_size,
+        enum GNUNET_PSYC_MessageFlags flags)
+{
+  return GNUNET_OK;
+}
+
+
+static int
+join (void *cls, const struct GNUNET_CRYPTO_EccPublicSignKey *slave_key,
+      const char *method_name,
+      size_t variable_count, const struct GNUNET_ENV_Modifier *variables,
+      const void *data, size_t data_size, struct GNUNET_PSYC_JoinHandle *jh)
+{
+  return GNUNET_OK;
+}
+
+
+void
+master_started (void *cls, uint64_t max_message_id)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Master started: %lu\n", max_message_id);
+}
+
+
+void
+slave_joined (void *cls, uint64_t max_message_id)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Slave joined: %lu\n", max_message_id);
+}
+
+
 /**
  * Main function of the test, run from scheduler.
  *
@@ -110,14 +162,28 @@ end ()
 static void
 #if DEBUG_SERVICE
 run (void *cls, char *const *args, const char *cfgfile,
-     const struct GNUNET_CONFIGURATION_Handle *cfg)
+     const struct GNUNET_CONFIGURATION_Handle *c)
 #else
 run (void *cls,
-     const struct GNUNET_CONFIGURATION_Handle *cfg,
+     const struct GNUNET_CONFIGURATION_Handle *c,
      struct GNUNET_TESTING_Peer *peer)
 #endif
 {
+  cfg = c;
   end_badly_task = GNUNET_SCHEDULER_add_delayed (TIMEOUT, &end_badly, NULL);
+
+  channel_key = GNUNET_CRYPTO_ecc_key_create ();
+  slave_key = GNUNET_CRYPTO_ecc_key_create ();
+
+  GNUNET_CRYPTO_ecc_key_get_public_for_signature (channel_key, &channel_pub_key);
+  GNUNET_CRYPTO_ecc_key_get_public_for_signature (slave_key, &slave_pub_key);
+
+  mst = GNUNET_PSYC_master_start (cfg, channel_key,
+                                  GNUNET_PSYC_CHANNEL_PRIVATE,
+                                  &method, &join, &master_started, NULL);
+
+  slv = GNUNET_PSYC_slave_join (cfg, &channel_pub_key, slave_key,
+                                &method, &join, &slave_joined, NULL);
 }