+check_client_host_enter (void *cls,
+ const struct HostEnterRequest *hr)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle a connecting client entering a place as host.
+ */
+static void
+handle_client_host_enter (void *cls,
+ const struct HostEnterRequest *hr)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ struct HostEnterRequest *
+ hreq = (struct HostEnterRequest *) GNUNET_copy_message (&hr->header);
+
+ uint8_t app_id_size = ntohs (hreq->header.size) - sizeof (*hreq);
+ const char *app_id = NULL;
+ uint16_t offset = GNUNET_STRINGS_buffer_tokenize ((const char *) &hreq[1],
+ app_id_size, 1, &app_id);
+ if (0 == offset || offset != app_id_size || app_id == NULL)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "offset = %u, app_id_size = %u, app_id = %s\n",
+ offset, app_id_size, app_id);
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ struct Host *hst = NULL;
+ struct Place *plc = NULL;
+ int ret = GNUNET_OK;
+
+ struct GNUNET_CRYPTO_EddsaPublicKey empty_pub_key;
+ memset (&empty_pub_key, 0, sizeof (empty_pub_key));
+
+ if (0 == memcmp (&hreq->place_pub_key, &empty_pub_key, sizeof (empty_pub_key)))
+ { // no public key set: create new private key & save the place
+ struct GNUNET_CRYPTO_EddsaPrivateKey *
+ place_key = GNUNET_CRYPTO_eddsa_key_create ();
+ hreq->place_key = *place_key;
+ GNUNET_CRYPTO_eddsa_key_get_public (place_key, &hreq->place_pub_key);
+ GNUNET_CRYPTO_eddsa_key_clear (place_key);
+ GNUNET_free (place_key);
+
+ app_place_save (app_id, (const struct PlaceEnterRequest *) hreq);
+ }
+
+ switch (host_enter (hreq, &hst))
+ {
+ case GNUNET_YES:
+ plc = c->place = &hst->place;
+ plc->host = hst;
+ break;
+
+ case GNUNET_NO:
+ {
+ plc = c->place = &hst->place;
+ plc->host = hst;
+ client_send_host_enter_ack (client, hst, GNUNET_OK);
+ break;
+ }
+ case GNUNET_SYSERR:
+ ret = GNUNET_SYSERR;
+ }
+
+ if (ret != GNUNET_SYSERR)
+ {
+
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p Client connected as host to place %s.\n",
+ hst, GNUNET_h2s (&plc->pub_key_hash));
+
+ struct ClientListItem *cli = GNUNET_new (struct ClientListItem);
+ cli->client = client;
+ GNUNET_CONTAINER_DLL_insert (plc->clients_head, plc->clients_tail, cli);
+ c->place = plc;
+ }
+
+ GNUNET_CRYPTO_eddsa_key_clear (&hreq->place_key);
+ GNUNET_free (hreq);
+
+ if (GNUNET_OK == ret)
+ GNUNET_SERVICE_client_continue (client);
+ else
+ GNUNET_SERVICE_client_drop (client);
+}
+
+
+/**
+ * Enter place as guest.
+ *
+ * @param greq
+ * Guest entry request.
+ * @param[out] ret_gst
+ * Returned Guest struct.
+ *
+ * @return #GNUNET_YES if the guest entered the place just now,
+ * #GNUNET_NO if the place is already entered,
+ * #GNUNET_SYSERR on error.
+ */
+static int
+guest_enter (const struct GuestEnterRequest *greq, struct Guest **ret_gst)
+{
+ int ret = GNUNET_NO;
+ uint16_t greq_size = ntohs (greq->header.size);
+
+ struct GNUNET_CRYPTO_EcdsaPublicKey ego_pub_key = greq->ego_pub_key;
+ struct GNUNET_HashCode ego_pub_hash;
+ GNUNET_CRYPTO_hash (&ego_pub_key, sizeof (ego_pub_key), &ego_pub_hash);
+ struct Ego *ego = GNUNET_CONTAINER_multihashmap_get (egos, &ego_pub_hash);
+
+ if (NULL == ego)
+ return GNUNET_SYSERR;
+
+ struct GNUNET_HashCode place_pub_hash;
+ GNUNET_CRYPTO_hash (&greq->place_pub_key, sizeof (greq->place_pub_key),
+ &place_pub_hash);
+
+ struct GNUNET_CONTAINER_MultiHashMap *
+ plc_gst = GNUNET_CONTAINER_multihashmap_get (place_guests, &place_pub_hash);
+ struct Guest *gst = NULL;
+
+ if (NULL != plc_gst)
+ gst = GNUNET_CONTAINER_multihashmap_get (plc_gst, &ego_pub_hash);
+
+ if (NULL == gst || NULL == gst->slave)
+ {
+ gst = GNUNET_new (struct Guest);
+ gst->origin = greq->origin;
+ gst->relay_count = ntohl (greq->relay_count);
+
+ uint16_t len;
+ uint16_t remaining = ntohs (greq->header.size) - sizeof (*greq);
+ const char *app_id = (const char *) &greq[1];
+ const char *p = app_id;
+
+ len = strnlen (app_id, remaining);
+ if (len == remaining)
+ {
+ GNUNET_free (gst);
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ p += len + 1;
+ remaining -= len + 1;
+
+ const struct GNUNET_PeerIdentity *relays = NULL;
+ uint16_t relay_size = gst->relay_count * sizeof (*relays);
+ if (remaining < relay_size)
+ {
+ GNUNET_free (gst);
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ if (0 < relay_size)
+ relays = (const struct GNUNET_PeerIdentity *) p;
+ p += relay_size;
+ remaining -= relay_size;
+
+ struct GNUNET_PSYC_Message *join_msg = NULL;
+ uint16_t join_msg_size = 0;
+
+ if (sizeof (struct GNUNET_MessageHeader) <= remaining)
+ {
+ join_msg = (struct GNUNET_PSYC_Message *) p;
+ join_msg_size = ntohs (join_msg->header.size);
+ p += join_msg_size;
+ remaining -= join_msg_size;
+ }
+ if (0 != remaining)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%zu + %u + %u != %u\n",
+ sizeof (*greq), relay_size, join_msg_size, greq_size);
+ GNUNET_free (gst);
+ GNUNET_break (0);
+ return GNUNET_SYSERR;
+ }
+ if (0 < relay_size)
+ {
+ gst->relays = GNUNET_malloc (relay_size);
+ GNUNET_memcpy (gst->relays, relays, relay_size);
+ }
+
+ gst->join_flags = ntohl (greq->flags);
+
+ struct Place *plc = &gst->place;
+ place_init (plc);
+ plc->is_host = GNUNET_NO;
+ plc->pub_key = greq->place_pub_key;
+ plc->pub_key_hash = place_pub_hash;
+ plc->ego_pub_key = ego_pub_key;
+ plc->ego_pub_hash = ego_pub_hash;
+ plc->ego_key = ego->key;
+
+ if (NULL == plc_gst)
+ {
+ plc_gst = GNUNET_CONTAINER_multihashmap_create (1, GNUNET_YES);
+ (void) GNUNET_CONTAINER_multihashmap_put (place_guests, &plc->pub_key_hash, plc_gst,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ }
+ (void) GNUNET_CONTAINER_multihashmap_put (plc_gst, &plc->ego_pub_hash, gst,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ (void) GNUNET_CONTAINER_multihashmap_put (guests, &plc->pub_key_hash, gst,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ gst->slave
+ = GNUNET_PSYC_slave_join (cfg, &plc->pub_key, &plc->ego_key,
+ gst->join_flags, &gst->origin,
+ gst->relay_count, gst->relays,
+ &psyc_recv_message, NULL,
+ &psyc_slave_connected,
+ &psyc_recv_join_dcsn,
+ gst, join_msg);
+ plc->channel = GNUNET_PSYC_slave_get_channel (gst->slave);
+ ret = GNUNET_YES;
+ }
+
+ if (NULL != ret_gst)
+ *ret_gst = gst;
+ return ret;
+}
+
+
+static int
+check_client_guest_enter (void *cls,
+ const struct GuestEnterRequest *greq)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle a connecting client entering a place as guest.
+ */
+static void
+handle_client_guest_enter (void *cls,
+ const struct GuestEnterRequest *greq)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ uint16_t remaining = ntohs (greq->header.size) - sizeof (*greq);
+ const char *app_id = NULL;
+ uint16_t offset = GNUNET_STRINGS_buffer_tokenize ((const char *) &greq[1],
+ remaining, 1, &app_id);
+ if (0 == offset)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ struct Guest *gst = NULL;
+ struct Place *plc = NULL;
+
+ switch (guest_enter (greq, &gst))
+ {
+ case GNUNET_YES:
+ plc = c->place = &gst->place;
+ plc->guest = gst;
+ app_place_save (app_id, (const struct PlaceEnterRequest *) greq);
+ break;
+
+ case GNUNET_NO:
+ {
+ plc = c->place = &gst->place;
+ plc->guest = gst;
+
+ struct GNUNET_PSYC_CountersResultMessage res;
+ res.header.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_GUEST_ENTER_ACK);
+ res.header.size = htons (sizeof (res));
+ res.result_code = htonl (GNUNET_OK);
+ res.max_message_id = GNUNET_htonll (plc->max_message_id);
+
+ client_send_msg (client, &res.header);
+ if (NULL != gst->join_dcsn)
+ client_send_msg (client, &gst->join_dcsn->header);
+
+ break;
+ }
+ case GNUNET_SYSERR:
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Client connected as guest to place %s.\n",
+ gst, GNUNET_h2s (&plc->pub_key_hash));
+
+ struct ClientListItem *cli = GNUNET_new (struct ClientListItem);
+ cli->client = client;
+ GNUNET_CONTAINER_DLL_insert (plc->clients_head, plc->clients_tail, cli);
+
+ c->place = plc;
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+struct GuestEnterByNameClosure
+{
+ struct Client *client;
+ char *app_id;
+ char *password;
+ struct GNUNET_CRYPTO_EcdsaPublicKey ego_pub_key;
+ struct GNUNET_MessageHeader *join_msg;
+};
+
+
+/**
+ * Result of a GNS name lookup for entering a place.
+ *
+ * @see GNUNET_SOCIAL_guest_enter_by_name
+ */
+static void
+gns_result_guest_enter (void *cls, uint32_t rd_count,
+ const struct GNUNET_GNSRECORD_Data *rd)
+{
+ struct GuestEnterByNameClosure *gcls = cls;
+ struct Client *c = gcls->client;
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p GNS result: %u records.\n",
+ c, rd_count);
+
+ const struct GNUNET_GNSRECORD_PlaceData *
+ rec = (const struct GNUNET_GNSRECORD_PlaceData *) rd->data;
+
+ if (0 == rd_count || rd->data_size < sizeof (*rec))
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (c->client);
+ return;
+ }
+
+ uint16_t relay_count = ntohl (rec->relay_count);
+ struct GNUNET_PeerIdentity *relays = NULL;
+
+ if (0 < relay_count)
+ {
+ if (rd->data_size == sizeof (*rec) + relay_count * sizeof (struct GNUNET_PeerIdentity))
+ {
+ relays = (struct GNUNET_PeerIdentity *) &rec[1];
+ }
+ else
+ {
+ relay_count = 0;
+ GNUNET_break_op (0);
+ }
+ }
+
+ uint16_t app_id_size = strlen (gcls->app_id) + 1;
+ uint16_t relay_size = relay_count * sizeof (*relays);
+ uint16_t join_msg_size = 0;
+ if (NULL != gcls->join_msg)
+ join_msg_size = ntohs (gcls->join_msg->size);
+ uint16_t greq_size = sizeof (struct GuestEnterRequest)
+ + app_id_size + relay_size + join_msg_size;
+ struct GuestEnterRequest *greq = GNUNET_malloc (greq_size);
+ greq->header.size = htons (greq_size);
+ greq->header.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_GUEST_ENTER);
+ greq->ego_pub_key = gcls->ego_pub_key;
+ greq->place_pub_key = rec->place_pub_key;
+ greq->origin = rec->origin;
+ greq->relay_count = rec->relay_count;
+
+ void *p = &greq[1];
+ GNUNET_memcpy (p, gcls->app_id, app_id_size);
+ p += app_id_size;
+ GNUNET_memcpy (p, relays, relay_size);
+ p += relay_size;
+ GNUNET_memcpy (p, gcls->join_msg, join_msg_size);
+
+ handle_client_guest_enter (c, greq);
+
+ GNUNET_free (gcls->app_id);
+ if (NULL != gcls->password)
+ GNUNET_free (gcls->password);
+ if (NULL != gcls->join_msg)
+ GNUNET_free (gcls->join_msg);
+ GNUNET_free (gcls);
+ GNUNET_free (greq);
+}
+
+
+static int
+check_client_guest_enter_by_name (void *cls,
+ const struct GuestEnterByNameRequest *greq)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle a connecting client entering a place as guest using a GNS address.
+ *
+ * Look up GNS address and generate a GuestEnterRequest from that.
+ */
+static void
+handle_client_guest_enter_by_name (void *cls,
+ const struct GuestEnterByNameRequest *greq)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ struct GuestEnterByNameClosure *gcls = GNUNET_malloc (sizeof (*gcls));
+ gcls->client = c;
+ gcls->ego_pub_key = greq->ego_pub_key;
+
+ const char *p = (const char *) &greq[1];
+ const char *app_id = NULL, *password = NULL, *gns_name = NULL;
+ uint16_t remaining = ntohs (greq->header.size) - sizeof (*greq);
+ uint16_t offset = GNUNET_STRINGS_buffer_tokenize (p, remaining, 3,
+ &app_id,
+ &gns_name,
+ &password);
+ p += offset;
+ remaining -= offset;
+
+ if (0 != offset && sizeof (*gcls->join_msg) <= remaining)
+ {
+ gcls->join_msg = GNUNET_copy_message ((struct GNUNET_MessageHeader *) p);
+ remaining -= ntohs (gcls->join_msg->size);
+ }
+
+ if (0 == offset || 0 != remaining)
+ {
+ if (NULL != gcls->join_msg)
+ GNUNET_free (gcls->join_msg);
+ GNUNET_free (gcls);
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ uint16_t app_id_size = strlen (app_id) + 1;
+ gcls->app_id = GNUNET_malloc (app_id_size);
+ GNUNET_memcpy (gcls->app_id, app_id, app_id_size);
+
+ uint16_t password_size = strlen (password);
+ if (0 < password_size++)
+ {
+ gcls->password = GNUNET_malloc (password_size);
+ GNUNET_memcpy (gcls->password, password, password_size);
+ }
+
+ GNUNET_GNS_lookup (gns, gns_name, &greq->ego_pub_key,
+ GNUNET_GNSRECORD_TYPE_PLACE, GNUNET_GNS_LO_DEFAULT,
+ NULL, gns_result_guest_enter, gcls);
+}
+
+
+void
+app_notify_place (struct GNUNET_MessageHeader *msg,
+ struct GNUNET_SERVICE_Client *client)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Sending place notification of type %u to client.\n",
+ client, ntohs (msg->type));
+
+ uint16_t msg_size = ntohs (msg->size);
+ struct AppPlaceMessage amsg;
+ amsg.header.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_APP_PLACE);
+ amsg.header.size = htons (sizeof (amsg));
+ // FIXME: also notify about not entered places
+ amsg.place_state = GNUNET_SOCIAL_PLACE_STATE_ENTERED;
+
+ switch (ntohs (msg->type))
+ {
+ case GNUNET_MESSAGE_TYPE_SOCIAL_HOST_ENTER:
+ if (msg_size < sizeof (struct HostEnterRequest))
+ return;
+ struct HostEnterRequest *hreq = (struct HostEnterRequest *) msg;
+ amsg.is_host = GNUNET_YES;
+ amsg.ego_pub_key = hreq->ego_pub_key;
+ amsg.place_pub_key = hreq->place_pub_key;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_SOCIAL_GUEST_ENTER:
+ if (msg_size < sizeof (struct GuestEnterRequest))
+ return;
+ struct GuestEnterRequest *greq = (struct GuestEnterRequest *) msg;
+ amsg.is_host = GNUNET_NO;
+ amsg.ego_pub_key = greq->ego_pub_key;
+ amsg.place_pub_key = greq->place_pub_key;
+ break;
+
+ default:
+ return;
+ }
+
+ client_send_msg (client, &amsg.header);
+}
+
+
+void
+app_notify_place_end (struct GNUNET_SERVICE_Client *client)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Sending end of place list notification to client\n",
+ client);
+
+ struct GNUNET_MessageHeader msg;
+ msg.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_APP_PLACE_END);
+ msg.size = htons (sizeof (msg));
+
+ client_send_msg (client, &msg);
+}
+
+
+void
+app_notify_ego (struct Ego *ego, struct GNUNET_SERVICE_Client *client)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Sending ego notification to client: %s\n",
+ client, ego->name);
+
+ size_t name_size = strlen (ego->name) + 1;
+ struct AppEgoMessage *emsg = GNUNET_malloc (sizeof (*emsg) + name_size);
+ emsg->header.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_APP_EGO);
+ emsg->header.size = htons (sizeof (*emsg) + name_size);
+
+ GNUNET_CRYPTO_ecdsa_key_get_public (&ego->key, &emsg->ego_pub_key);
+ GNUNET_memcpy (&emsg[1], ego->name, name_size);
+
+ client_send_msg (client, &emsg->header);
+ GNUNET_free (emsg);
+}
+
+
+void
+app_notify_ego_end (struct GNUNET_SERVICE_Client *client)
+{
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Sending end of ego list notification to client\n",
+ client);
+
+ struct GNUNET_MessageHeader msg;
+ msg.type = htons (GNUNET_MESSAGE_TYPE_SOCIAL_APP_EGO_END);
+ msg.size = htons (sizeof (msg));
+
+ client_send_msg (client, &msg);
+}
+
+
+int
+app_place_entry_notify (void *cls, const struct GNUNET_HashCode *key, void *value)
+{
+ struct GNUNET_MessageHeader *
+ msg = GNUNET_CONTAINER_multihashmap_get (places, key);
+ if (NULL != msg)
+ app_notify_place (msg, cls);
+ return GNUNET_YES;
+}
+
+
+int
+ego_entry (void *cls, const struct GNUNET_HashCode *key, void *value)
+{
+ app_notify_ego (value, cls);
+ return GNUNET_YES;
+}
+
+
+static int
+check_client_app_connect (void *cls,
+ const struct AppConnectRequest *creq)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle application connection.
+ */
+static void
+handle_client_app_connect (void *cls,
+ const struct AppConnectRequest *creq)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ uint8_t app_id_size = ntohs (creq->header.size) - sizeof (*creq);
+ const char *app_id = NULL;
+ uint16_t offset = GNUNET_STRINGS_buffer_tokenize ((const char *) &creq[1],
+ app_id_size, 1, &app_id);
+ if (0 == offset || offset != app_id_size)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ struct GNUNET_HashCode app_id_hash;
+ GNUNET_CRYPTO_hash (app_id, app_id_size, &app_id_hash);
+
+ GNUNET_CONTAINER_multihashmap_iterate (egos, ego_entry, client);
+ app_notify_ego_end (client);
+
+ struct GNUNET_CONTAINER_MultiHashMap *
+ app_places = GNUNET_CONTAINER_multihashmap_get (apps_places, &app_id_hash);
+ if (NULL != app_places)
+ GNUNET_CONTAINER_multihashmap_iterate (app_places, app_place_entry_notify, client);
+ app_notify_place_end (client);
+
+ struct ClientListItem *cli = GNUNET_new (struct ClientListItem);
+ cli->client = client;
+ struct Application *app = GNUNET_CONTAINER_multihashmap_get (apps,
+ &app_id_hash);
+ if (NULL == app) {
+ app = GNUNET_malloc (sizeof (*app));
+ (void) GNUNET_CONTAINER_multihashmap_put (apps, &app_id_hash, app,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_FAST);
+ }
+ GNUNET_CONTAINER_DLL_insert (app->clients_head, app->clients_tail, cli);
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Application %s connected.\n", app, app_id);
+
+ c->app_id = GNUNET_malloc (app_id_size);
+ GNUNET_memcpy (c->app_id, app_id, app_id_size);
+
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * Handle application detach request.
+ */
+static void
+handle_client_app_detach (void *cls,
+ const struct AppDetachRequest *req)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+
+ int ret = app_place_remove (c->app_id, &req->ego_pub_key, &req->place_pub_key);
+ client_send_result (client, req->op_id, ret, NULL, 0);
+
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+int
+app_places_entry_remove (void *cls, const struct GNUNET_HashCode *key, void *value)
+{
+ struct Place *plc = cls;
+ const char *app_id = value;
+ app_place_remove (app_id, &plc->ego_pub_key, &plc->pub_key);
+ return GNUNET_YES;
+}
+
+
+/**
+ * Handle application leave request.
+ */
+static void
+handle_client_place_leave (void *cls,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+ struct Place *plc = c->place;
+ if (NULL == plc)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ /* FIXME: remove all app subscriptions and leave this place */
+
+ struct GNUNET_CONTAINER_MultiHashMap *
+ place_apps = GNUNET_CONTAINER_multihashmap_get (places_apps, &plc->pub_key_hash);
+ if (NULL != place_apps)
+ {
+ GNUNET_CONTAINER_multihashmap_iterate (place_apps, app_places_entry_remove, plc);
+ }
+
+ /* FIXME: disconnect from the network, but keep local connection for history access */
+
+ /* Disconnect all clients connected to the place */
+ struct ClientListItem *cli = plc->clients_head, *next;
+ while (NULL != cli)
+ {
+ GNUNET_CONTAINER_DLL_remove (plc->clients_head, plc->clients_tail, cli);
+ GNUNET_SERVICE_client_drop (cli->client);
+ next = cli->next;
+ GNUNET_free (cli);
+ cli = next;
+ }
+
+ if (GNUNET_YES != plc->is_disconnected)
+ {
+ plc->is_disconnected = GNUNET_YES;
+ if (NULL != plc->tmit_msgs_head)
+ { /* Send pending messages to PSYC before cleanup. */
+ psyc_transmit_message (plc);
+ }
+ else
+ {
+ cleanup_place (plc);
+ }
+ }
+}
+
+
+struct JoinDecisionClosure
+{
+ int32_t is_admitted;
+ struct GNUNET_PSYC_Message *msg;
+};
+
+
+/**
+ * Iterator callback for responding to join requests.
+ */
+static int
+psyc_send_join_decision (void *cls, const struct GNUNET_HashCode *pub_key_hash,
+ void *value)
+{
+ struct JoinDecisionClosure *jcls = cls;
+ struct GNUNET_PSYC_JoinHandle *jh = value;
+ // FIXME: add relays
+ GNUNET_PSYC_join_decision (jh, jcls->is_admitted, 0, NULL, jcls->msg);
+ return GNUNET_YES;
+}
+
+
+static int
+check_client_join_decision (void *cls,
+ const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle an entry decision from a host client.
+ */
+static void
+handle_client_join_decision (void *cls,
+ const struct GNUNET_PSYC_JoinDecisionMessage *dcsn)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+ struct Place *plc = c->place;
+ if (NULL == plc || GNUNET_YES != plc->is_host)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+ struct Host *hst = plc->host;
+
+ struct JoinDecisionClosure jcls;
+ jcls.is_admitted = ntohl (dcsn->is_admitted);
+ jcls.msg
+ = (sizeof (*dcsn) + sizeof (*jcls.msg) <= ntohs (dcsn->header.size))
+ ? (struct GNUNET_PSYC_Message *) &dcsn[1]
+ : NULL;
+
+ struct GNUNET_HashCode slave_pub_hash;
+ GNUNET_CRYPTO_hash (&dcsn->slave_pub_key, sizeof (dcsn->slave_pub_key),
+ &slave_pub_hash);
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Got join decision (%d) from client for place %s..\n",
+ hst, jcls.is_admitted, GNUNET_h2s (&plc->pub_key_hash));
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p ..and slave %s.\n",
+ hst, GNUNET_h2s (&slave_pub_hash));
+
+ GNUNET_CONTAINER_multihashmap_get_multiple (hst->join_reqs, &slave_pub_hash,
+ &psyc_send_join_decision, &jcls);
+ GNUNET_CONTAINER_multihashmap_remove_all (hst->join_reqs, &slave_pub_hash);
+
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * Send acknowledgement to a client.
+ *
+ * Sent after a message fragment has been passed on to multicast.
+ *
+ * @param plc The place struct for the client.
+ */
+static void
+send_message_ack (struct Place *plc, struct GNUNET_SERVICE_Client *client)
+{
+ struct GNUNET_MessageHeader res;
+ res.size = htons (sizeof (res));
+ res.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK);
+ client_send_msg (client, &res);
+}
+
+
+/**
+ * Proceed to the next message part in the transmission queue.
+ *
+ * @param plc
+ * Place where the transmission is going on.
+ * @param tmit_msg
+ * Currently transmitted message.
+ * @param tmit_frag
+ * Currently transmitted message fragment.
+ *
+ * @return @a tmit_frag, or NULL if reached the end of fragment.
+ */
+static struct FragmentTransmitQueue *
+psyc_transmit_queue_next_part (struct Place *plc,
+ struct MessageTransmitQueue *tmit_msg,
+ struct FragmentTransmitQueue *tmit_frag)
+{
+ uint16_t psize = ntohs (tmit_frag->next_part->size);
+ if ((char *) tmit_frag->next_part + psize - ((char *) &tmit_frag[1])
+ < tmit_frag->size)
+ {
+ tmit_frag->next_part
+ = (struct GNUNET_MessageHeader *) ((char *) tmit_frag->next_part + psize);
+ }
+ else /* Reached end of current fragment. */
+ {
+ if (NULL != tmit_frag->client)
+ send_message_ack (plc, tmit_frag->client);
+ GNUNET_CONTAINER_DLL_remove (tmit_msg->frags_head, tmit_msg->frags_tail, tmit_frag);
+ GNUNET_free (tmit_frag);
+ tmit_frag = NULL;
+ }
+ return tmit_frag;
+}
+
+
+/**
+ * Proceed to next message in transmission queue.
+ *
+ * @param plc
+ * Place where the transmission is going on.
+ * @param tmit_msg
+ * Currently transmitted message.
+ *
+ * @return The next message in queue, or NULL if queue is empty.
+ */
+static struct MessageTransmitQueue *
+psyc_transmit_queue_next_msg (struct Place *plc,
+ struct MessageTransmitQueue *tmit_msg)
+{
+ GNUNET_CONTAINER_DLL_remove (plc->tmit_msgs_head, plc->tmit_msgs_tail, tmit_msg);
+ GNUNET_free (tmit_msg);
+ return plc->tmit_msgs_head;
+}
+
+
+/**
+ * Callback for data transmission to PSYC.
+ */
+static int
+psyc_transmit_notify_data (void *cls, uint16_t *data_size, void *data)
+{
+ struct Place *plc = cls;
+ struct MessageTransmitQueue *tmit_msg = plc->tmit_msgs_head;
+ GNUNET_assert (NULL != tmit_msg);
+ struct FragmentTransmitQueue *tmit_frag = tmit_msg->frags_head;
+ if (NULL == tmit_frag)
+ { /* Rest of the message have not arrived yet, pause transmission */
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+ struct GNUNET_MessageHeader *pmsg = tmit_frag->next_part;
+ if (NULL == pmsg)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_data: nothing to send.\n", plc);
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_data()\n", plc);
+ GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg);
+
+ uint16_t ptype = ntohs (pmsg->type);
+ uint16_t pdata_size = ntohs (pmsg->size) - sizeof (*pmsg);
+ int ret;
+
+ switch (ptype)
+ {
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+ if (*data_size < pdata_size)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_data: buffer size too small for data.\n", plc);
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_data: sending %u bytes.\n",
+ plc, pdata_size);
+
+ *data_size = pdata_size;
+ GNUNET_memcpy (data, &pmsg[1], *data_size);
+ ret = GNUNET_NO;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+ *data_size = 0;
+ ret = GNUNET_YES;
+ break;
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+ *data_size = 0;
+ ret = GNUNET_SYSERR;
+ break;
+
+ default:
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p psyc_transmit_notify_data: unexpected message part of type %u.\n",
+ plc, ptype);
+ ret = GNUNET_SYSERR;
+ }
+
+ if (GNUNET_SYSERR == ret && GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL != ptype)
+ {
+ *data_size = 0;
+ tmit_msg = psyc_transmit_queue_next_msg (plc, tmit_msg);
+ plc->is_disconnected = GNUNET_YES;
+ GNUNET_SERVICE_client_drop (tmit_frag->client);
+ GNUNET_SCHEDULER_add_now (&cleanup_place, plc);
+ return ret;
+ }
+ else
+ {
+ tmit_frag = psyc_transmit_queue_next_part (plc, tmit_msg, tmit_frag);
+ if (NULL != tmit_frag)
+ {
+ struct GNUNET_MessageHeader *pmsg = tmit_frag->next_part;
+ ptype = ntohs (pmsg->type);
+ switch (ptype)
+ {
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+ ret = GNUNET_YES;
+ break;
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+ ret = GNUNET_SYSERR;
+ break;
+ }
+ switch (ptype)
+ {
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+ tmit_frag = psyc_transmit_queue_next_part (plc, tmit_msg, tmit_frag);
+ }
+ }
+
+ if (NULL == tmit_msg->frags_head
+ && GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END <= ptype)
+ { /* Reached end of current message. */
+ tmit_msg = psyc_transmit_queue_next_msg (plc, tmit_msg);
+ }
+ }
+
+ if (ret != GNUNET_NO)
+ {
+ if (NULL != tmit_msg)
+ {
+ psyc_transmit_message (plc);
+ }
+ else if (GNUNET_YES == plc->is_disconnected)
+ {
+ /* FIXME: handle partial message (when still in_transmit) */
+ cleanup_place (plc);
+ }
+ }
+ return ret;
+}
+
+
+/**
+ * Callback for modifier transmission to PSYC.
+ */
+static int
+psyc_transmit_notify_mod (void *cls, uint16_t *data_size, void *data,
+ uint8_t *oper, uint32_t *full_value_size)
+{
+ struct Place *plc = cls;
+ struct MessageTransmitQueue *tmit_msg = plc->tmit_msgs_head;
+ GNUNET_assert (NULL != tmit_msg);
+ struct FragmentTransmitQueue *tmit_frag = tmit_msg->frags_head;
+ if (NULL == tmit_frag)
+ { /* Rest of the message have not arrived yet, pause transmission */
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+ struct GNUNET_MessageHeader *pmsg = tmit_frag->next_part;
+ if (NULL == pmsg)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_mod: nothing to send.\n", plc);
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_mod()\n", plc);
+ GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg);
+
+ uint16_t ptype = ntohs (pmsg->type);
+ int ret;
+
+ switch (ptype)
+ {
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER:
+ {
+ if (NULL == oper)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p psyc_transmit_notify_mod: oper is NULL.\n", plc);
+ ret = GNUNET_SYSERR;
+ break;
+ }
+ struct GNUNET_PSYC_MessageModifier *
+ pmod = (struct GNUNET_PSYC_MessageModifier *) tmit_frag->next_part;
+ uint16_t mod_size = ntohs (pmod->header.size) - sizeof (*pmod);
+
+ if (*data_size < mod_size)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_mod: buffer size too small for data.\n", plc);
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+
+ *full_value_size = ntohl (pmod->value_size);
+ *oper = pmod->oper;
+ *data_size = mod_size;
+ GNUNET_memcpy (data, &pmod[1], mod_size);
+ ret = GNUNET_NO;
+ break;
+ }
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT:
+ {
+ if (NULL != oper)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p psyc_transmit_notify_mod: oper is not NULL.\n", plc);
+ ret = GNUNET_SYSERR;
+ break;
+ }
+ uint16_t mod_size = ntohs (pmsg->size) - sizeof (*pmsg);
+ if (*data_size < mod_size)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_mod: buffer size too small for data.\n", plc);
+ *data_size = 0;
+ return GNUNET_NO;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p psyc_transmit_notify_mod: sending %u bytes.\n", plc, mod_size);
+
+ *data_size = mod_size;
+ GNUNET_memcpy (data, &pmsg[1], *data_size);
+ ret = GNUNET_NO;
+ break;
+ }
+
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA:
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END:
+ case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL:
+ *data_size = 0;
+ ret = GNUNET_YES;
+ break;
+
+ default:
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p psyc_transmit_notify_mod: unexpected message part of type %u.\n",
+ plc, ptype);
+ ret = GNUNET_SYSERR;
+ }
+
+ if (GNUNET_SYSERR == ret)
+ {
+ *data_size = 0;
+ ret = GNUNET_SYSERR;
+ tmit_msg = psyc_transmit_queue_next_msg (plc, tmit_msg);
+ plc->is_disconnected = GNUNET_YES;
+ GNUNET_SERVICE_client_drop (tmit_frag->client);
+ GNUNET_SCHEDULER_add_now (&cleanup_place, plc);
+ }
+ else
+ {
+ if (GNUNET_YES != ret)
+ psyc_transmit_queue_next_part (plc, tmit_msg, tmit_frag);
+
+ if (NULL == tmit_msg->frags_head
+ && GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END <= ptype)
+ { /* Reached end of current message. */
+ tmit_msg = psyc_transmit_queue_next_msg (plc, tmit_msg);
+ }
+ }
+ return ret;
+}
+
+/**
+ * Callback for data transmission from a host to PSYC.
+ */
+static int
+host_transmit_notify_data (void *cls, uint16_t *data_size, void *data)
+{
+ int ret = psyc_transmit_notify_data (cls, data_size, data);
+
+ if (GNUNET_NO != ret)
+ {
+ struct Host *hst = cls;
+ hst->tmit_handle = NULL;
+ }
+ return ret;
+}
+
+
+/**
+ * Callback for the transmit functions of multicast.
+ */
+static int
+guest_transmit_notify_data (void *cls, uint16_t *data_size, void *data)
+{
+ int ret = psyc_transmit_notify_data (cls, data_size, data);
+
+ if (GNUNET_NO != ret)
+ {
+ struct Guest *gst = cls;
+ gst->tmit_handle = NULL;
+ }
+ return ret;
+}
+
+
+/**
+ * Callback for modifier transmission from a host to PSYC.
+ */
+static int
+host_transmit_notify_mod (void *cls, uint16_t *data_size, void *data,
+ uint8_t *oper, uint32_t *full_value_size)
+{
+ int ret = psyc_transmit_notify_mod (cls, data_size, data,
+ oper, full_value_size);
+ if (GNUNET_SYSERR == ret)
+ {
+ struct Host *hst = cls;
+ hst->tmit_handle = NULL;
+ }
+ return ret;
+}
+
+
+/**
+ * Callback for modifier transmission from a guest to PSYC.
+ */
+static int
+guest_transmit_notify_mod (void *cls, uint16_t *data_size, void *data,
+ uint8_t *oper, uint32_t *full_value_size)
+{
+ int ret = psyc_transmit_notify_mod (cls, data_size, data,
+ oper, full_value_size);
+ if (GNUNET_SYSERR == ret)
+ {
+ struct Guest *gst = cls;
+ gst->tmit_handle = NULL;
+ }
+ return ret;
+}
+
+
+/**
+ * Get method part of next message from transmission queue.
+ *
+ * @param plc
+ * Place
+ *
+ * @return #GNUNET_OK on success
+ * #GNUNET_NO if there are no more messages in queue.
+ * #GNUNET_SYSERR if the next message is malformed.
+ */
+static struct GNUNET_PSYC_MessageMethod *
+psyc_transmit_queue_next_method (struct Place *plc)
+{
+ struct MessageTransmitQueue *tmit_msg = plc->tmit_msgs_head;
+ if (NULL == tmit_msg)
+ return GNUNET_NO;
+
+ struct FragmentTransmitQueue *tmit_frag = tmit_msg->frags_head;
+ if (NULL == tmit_frag)
+ {
+ GNUNET_break (0);
+ return GNUNET_NO;
+ }
+
+ struct GNUNET_MessageHeader *pmsg = tmit_frag->next_part;
+ if (NULL == pmsg
+ || GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD != ntohs (pmsg->type))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p psyc_transmit_queue_next_method: unexpected message part of type %u.\n",
+ plc, NULL != pmsg ? ntohs (pmsg->type) : 0);
+ GNUNET_break (0);
+ return NULL;
+ }
+
+ uint16_t psize = ntohs (pmsg->size);
+ struct GNUNET_PSYC_MessageMethod *
+ pmeth = (struct GNUNET_PSYC_MessageMethod *) GNUNET_copy_message (pmsg);
+
+ if (psize < sizeof (*pmeth) + 1 || '\0' != *((char *) pmeth + psize - 1))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p psyc_transmit_queue_next_method: invalid method name.\n",
+ plc);
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%zu <= %u || NUL != %u\n",
+ sizeof (*pmeth), psize, *((char *) pmeth + psize - 1));
+ GNUNET_break (0);
+ GNUNET_free (pmeth);
+ return NULL;
+ }
+
+ psyc_transmit_queue_next_part (plc, tmit_msg, tmit_frag);
+ return pmeth;
+}
+
+
+/**
+ * Transmit the next message in queue from the host to the PSYC channel.
+ */
+static int
+psyc_master_transmit_message (struct Host *hst)
+{
+ struct Place *plc = &hst->place;
+
+ if (NULL == hst->tmit_handle)
+ {
+ struct GNUNET_PSYC_MessageMethod *
+ pmeth = psyc_transmit_queue_next_method (plc);
+ if (NULL == pmeth)
+ return GNUNET_SYSERR;
+
+ hst->tmit_handle = (void *) &hst->tmit_handle;
+ struct GNUNET_PSYC_MasterTransmitHandle *
+ tmit_handle = GNUNET_PSYC_master_transmit (hst->master, (const char *) &pmeth[1],
+ &host_transmit_notify_mod,
+ &host_transmit_notify_data, hst,
+ pmeth->flags);
+ if (NULL != hst->tmit_handle)
+ hst->tmit_handle = tmit_handle;
+ GNUNET_free (pmeth);
+ }
+ else
+ {
+ GNUNET_PSYC_master_transmit_resume (hst->tmit_handle);
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Transmit the next message in queue from a guest to the PSYC channel.
+ */
+static int
+psyc_slave_transmit_message (struct Guest *gst)
+{
+ struct Place *plc = &gst->place;
+
+ if (NULL == gst->tmit_handle)
+ {
+ struct GNUNET_PSYC_MessageMethod *
+ pmeth = psyc_transmit_queue_next_method (plc);
+ if (NULL == pmeth)
+ return GNUNET_SYSERR;
+
+ gst->tmit_handle = (void *) &gst->tmit_handle;
+ struct GNUNET_PSYC_SlaveTransmitHandle *
+ tmit_handle = GNUNET_PSYC_slave_transmit (gst->slave, (const char *) &pmeth[1],
+ &guest_transmit_notify_mod,
+ &guest_transmit_notify_data, gst,
+ pmeth->flags);
+ if (NULL != gst->tmit_handle)
+ gst->tmit_handle = tmit_handle;
+ GNUNET_free (pmeth);
+ }
+ else
+ {
+ GNUNET_PSYC_slave_transmit_resume (gst->tmit_handle);
+ }
+ return GNUNET_OK;
+}
+
+
+/**
+ * Transmit a message to PSYC.
+ */
+static int
+psyc_transmit_message (struct Place *plc)
+{
+ return
+ (plc->is_host)
+ ? psyc_master_transmit_message ((struct Host *) plc)
+ : psyc_slave_transmit_message ((struct Guest *) plc);
+}
+
+
+/**
+ * Queue message parts for sending to PSYC.
+ *
+ * @param plc Place to send to.
+ * @param client Client the message originates from.
+ * @param data_size Size of @a data.
+ * @param data Concatenated message parts.
+ * @param first_ptype First message part type in @a data.
+ * @param last_ptype Last message part type in @a data.
+ */
+static struct MessageTransmitQueue *
+psyc_transmit_queue_message (struct Place *plc,
+ struct GNUNET_SERVICE_Client *client,
+ size_t data_size,
+ const void *data,
+ uint16_t first_ptype, uint16_t last_ptype,
+ struct MessageTransmitQueue *tmit_msg)
+{
+ if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD == first_ptype)
+ {
+ tmit_msg = GNUNET_malloc (sizeof (*tmit_msg));
+ GNUNET_CONTAINER_DLL_insert_tail (plc->tmit_msgs_head, plc->tmit_msgs_tail, tmit_msg);
+ }
+ else if (NULL == tmit_msg)
+ {
+ return NULL;
+ }
+
+ struct FragmentTransmitQueue *
+ tmit_frag = GNUNET_malloc (sizeof (*tmit_frag) + data_size);
+ GNUNET_memcpy (&tmit_frag[1], data, data_size);
+ tmit_frag->next_part = (struct GNUNET_MessageHeader *) &tmit_frag[1];
+ tmit_frag->client = client;
+ tmit_frag->size = data_size;
+
+ GNUNET_CONTAINER_DLL_insert_tail (tmit_msg->frags_head, tmit_msg->frags_tail, tmit_frag);
+ tmit_msg->client = client;
+ return tmit_msg;
+}
+
+
+/**
+ * Cancel transmission of current message to PSYC.
+ *
+ * @param plc Place to send to.
+ * @param client Client the message originates from.
+ */
+static void
+psyc_transmit_cancel (struct Place *plc, struct GNUNET_SERVICE_Client *client)
+{
+ uint16_t type = GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL;
+
+ struct GNUNET_MessageHeader msg;
+ msg.size = htons (sizeof (msg));
+ msg.type = htons (type);
+
+ psyc_transmit_queue_message (plc, client, sizeof (msg), &msg, type, type, NULL);
+ psyc_transmit_message (plc);
+
+ /* FIXME: cleanup */
+}
+
+
+static int
+check_client_psyc_message (void *cls,
+ const struct GNUNET_MessageHeader *msg)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Handle an incoming message from a client, to be transmitted to the place.
+ */
+static void
+handle_client_psyc_message (void *cls,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+ struct Place *plc = c->place;
+ if (NULL == plc)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ int ret = GNUNET_SYSERR;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Received message from client.\n", plc);
+ GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, msg);
+
+ if (GNUNET_YES != plc->is_ready)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+ "%p Place is not ready yet, disconnecting client.\n", plc);
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ uint16_t size = ntohs (msg->size);
+ uint16_t psize = size - sizeof (*msg);
+ if (psize < sizeof (struct GNUNET_MessageHeader)
+ || GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < psize)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p Received message with invalid payload size (%u) from client.\n",
+ plc, psize);
+ GNUNET_break (0);
+ psyc_transmit_cancel (plc, client);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ uint16_t first_ptype = 0, last_ptype = 0;
+ if (GNUNET_SYSERR
+ == GNUNET_PSYC_receive_check_parts (psize, (const char *) &msg[1],
+ &first_ptype, &last_ptype))
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p Received invalid message part from client.\n", plc);
+ GNUNET_break (0);
+ psyc_transmit_cancel (plc, client);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Received message with first part type %u and last part type %u.\n",
+ plc, first_ptype, last_ptype);
+
+ c->tmit_msg
+ = psyc_transmit_queue_message (plc, client, psize, &msg[1],
+ first_ptype, last_ptype, c->tmit_msg);
+ if (NULL != c->tmit_msg)
+ {
+ if (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END <= last_ptype)
+ c->tmit_msg = NULL;
+ ret = psyc_transmit_message (plc);
+ }
+
+ if (GNUNET_OK != ret)
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p Received invalid message part from client.\n", plc);
+ GNUNET_break (0);
+ psyc_transmit_cancel (plc, client);
+ ret = GNUNET_SYSERR;
+ }
+
+ if (GNUNET_OK == ret)
+ GNUNET_SERVICE_client_continue (client);
+ else
+ GNUNET_SERVICE_client_drop (client);
+}
+
+
+/**
+ * A historic message arrived from PSYC.
+ */
+static void
+psyc_recv_history_message (void *cls, const struct GNUNET_PSYC_MessageHeader *msg)
+{
+ struct OperationClosure *opcls = cls;
+ struct Client *c = opcls->client;
+ struct Place *plc = c->place;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Received historic message #%" PRId64 " (flags: %x)\n",
+ plc, GNUNET_ntohll (msg->message_id), ntohl (msg->flags));
+
+ uint16_t size = ntohs (msg->header.size);
+
+ struct GNUNET_OperationResultMessage *
+ res = GNUNET_malloc (sizeof (*res) + size);
+ res->header.size = htons (sizeof (*res) + size);
+ res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT);
+ res->op_id = opcls->op_id;
+ res->result_code = GNUNET_htonll (GNUNET_OK);
+
+ GNUNET_memcpy (&res[1], msg, size);
+
+ /** @todo FIXME: send only to requesting client */
+ place_send_msg (plc, &res->header);
+
+ GNUNET_free (res);
+}
+
+
+/**
+ * Result of message history replay from PSYC.
+ */
+static void
+psyc_recv_history_result (void *cls, int64_t result,
+ const void *err_msg, uint16_t err_msg_size)
+{
+ struct OperationClosure *opcls = cls;
+ struct Client *c = opcls->client;
+ struct Place *plc = c->place;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p History replay #%" PRIu64 ": "
+ "PSYCstore returned %" PRId64 " (%.*s)\n",
+ plc, GNUNET_ntohll (opcls->op_id), result,
+ err_msg_size, (const char *) err_msg);
+
+ // FIXME: place might have been destroyed
+ client_send_result (c->client, opcls->op_id, result, err_msg, err_msg_size);
+}
+
+
+static int
+check_client_history_replay (void *cls,
+ const struct GNUNET_PSYC_HistoryRequestMessage *req)
+{
+ return GNUNET_OK;
+}
+
+
+/**
+ * Client requests channel history.
+ */
+static void
+handle_client_history_replay (void *cls,
+ const struct GNUNET_PSYC_HistoryRequestMessage *req)
+{
+ struct Client *c = cls;
+ struct GNUNET_SERVICE_Client *client = c->client;
+ struct Place *plc = c->place;
+ if (NULL == plc)
+ {
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ uint16_t size = ntohs (req->header.size);
+ const char *method_prefix = (const char *) &req[1];
+
+ if (size < sizeof (*req) + 1
+ || '\0' != method_prefix[size - sizeof (*req) - 1])
+ {
+ GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+ "%p History replay #%" PRIu64 ": "
+ "invalid method prefix. size: %u < %zu?\n",
+ plc, GNUNET_ntohll (req->op_id), size, sizeof (*req) + 1);
+ GNUNET_break (0);
+ GNUNET_SERVICE_client_drop (client);
+ return;
+ }
+
+ struct OperationClosure *opcls = GNUNET_malloc (sizeof (*opcls));
+ opcls->client = c;
+ opcls->op_id = req->op_id;
+ opcls->flags = ntohl (req->flags);
+
+ if (0 == req->message_limit)
+ GNUNET_PSYC_channel_history_replay (plc->channel,
+ GNUNET_ntohll (req->start_message_id),
+ GNUNET_ntohll (req->end_message_id),
+ method_prefix, opcls->flags,
+ psyc_recv_history_message, NULL,
+ psyc_recv_history_result, opcls);
+ else
+ GNUNET_PSYC_channel_history_replay_latest (plc->channel,
+ GNUNET_ntohll (req->message_limit),
+ method_prefix, opcls->flags,
+ psyc_recv_history_message, NULL,
+ psyc_recv_history_result, opcls);
+
+ GNUNET_SERVICE_client_continue (client);
+}
+
+
+/**
+ * A state variable part arrived from PSYC.
+ */
+void
+psyc_recv_state_var (void *cls,
+ const struct GNUNET_MessageHeader *mod,
+ const char *name,
+ const void *value,
+ uint32_t value_size,
+ uint32_t full_value_size)
+{
+ struct OperationClosure *opcls = cls;
+ struct Client *c = opcls->client;
+ struct Place *plc = c->place;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p Received state variable %s from PSYC\n",
+ plc, name);
+
+ uint16_t size = ntohs (mod->size);
+
+ struct GNUNET_OperationResultMessage *
+ res = GNUNET_malloc (sizeof (*res) + size);
+ res->header.size = htons (sizeof (*res) + size);
+ res->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT);
+ res->op_id = opcls->op_id;
+ res->result_code = GNUNET_htonll (GNUNET_OK);
+
+ GNUNET_memcpy (&res[1], mod, size);
+
+ /** @todo FIXME: send only to requesting client */
+ place_send_msg (plc, &res->header);
+
+ GNUNET_free (res);
+}
+
+
+/**
+ * Result of retrieving state variable from PSYC.
+ */
+static void
+psyc_recv_state_result (void *cls, int64_t result,
+ const void *err_msg, uint16_t err_msg_size)
+{
+ struct OperationClosure *opcls = cls;
+ struct Client *c = opcls->client;
+ struct Place *plc = c->place;
+
+ GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+ "%p State get #%" PRIu64 ": "
+ "PSYCstore returned %" PRId64 " (%.*s)\n",
+ plc, GNUNET_ntohll (opcls->op_id), result,
+ err_msg_size, (const char *) err_msg);
+
+ // FIXME: place might have been destroyed
+ client_send_result (c->client, opcls->op_id, result, err_msg, err_msg_size);
+}
+
+
+static int
+check_client_state_get (void *cls,
+ const struct GNUNET_PSYC_StateRequestMessage *req)