diff options
Diffstat (limited to 'src/psyc/psyc_api.c')
| -rw-r--r-- | src/psyc/psyc_api.c | 1295 |
1 files changed, 221 insertions, 1074 deletions
diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c index 7ec9d21b72..bfb6f43fb2 100644 --- a/src/psyc/psyc_api.c +++ b/src/psyc/psyc_api.c @@ -37,29 +37,11 @@ #include "gnunet_env_lib.h" #include "gnunet_multicast_service.h" #include "gnunet_psyc_service.h" +#include "gnunet_psyc_util_lib.h" #include "psyc.h" #define LOG(kind,...) GNUNET_log_from (kind, "psyc-api",__VA_ARGS__) -struct MessageQueue -{ - struct MessageQueue *prev; - struct MessageQueue *next; - /* Followed by struct GNUNET_MessageHeader msg */ -}; - - -/** - * Handle for a pending PSYC transmission operation. - */ -struct GNUNET_PSYC_ChannelTransmitHandle -{ - struct GNUNET_PSYC_Channel *ch; - GNUNET_PSYC_TransmitNotifyModifier notify_mod; - GNUNET_PSYC_TransmitNotifyData notify_data; - void *notify_cls; - enum MessageState state; -}; /** * Handle to access PSYC channel operations for both the master and slaves. @@ -67,109 +49,29 @@ struct GNUNET_PSYC_ChannelTransmitHandle struct GNUNET_PSYC_Channel { /** - * Transmission handle; - */ - struct GNUNET_PSYC_ChannelTransmitHandle tmit; - - /** * 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. + * Client connection to the service. */ - struct GNUNET_CLIENT_TransmitHandle *th; + struct GNUNET_CLIENT_MANAGER_Connection *client; /** - * Head of messages to transmit to the service. - */ - struct MessageQueue *tmit_head; - - /** - * Tail of operations to transmit to the service. + * Transmission handle; */ - struct MessageQueue *tmit_tail; + struct GNUNET_PSYC_TransmitHandle *tmit; /** - * Message currently being transmitted to the service. + * Receipt handle; */ - struct MessageQueue *tmit_msg; + struct GNUNET_PSYC_ReceiveHandle *recv; /** * 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; - - /** - * Message part callback. - */ - GNUNET_PSYC_MessageCallback message_cb; - - /** - * Message part callback for historic message. - */ - GNUNET_PSYC_MessageCallback hist_message_cb; - - /** - * Closure for @a message_cb. - */ - void *cb_cls; - - /** - * ID of the message being received from the PSYC service. - */ - uint64_t recv_message_id; - - /** - * Public key of the slave from which a message is being received. - */ - struct GNUNET_CRYPTO_EddsaPublicKey recv_slave_key; - - /** - * State of the currently being received message from the PSYC service. - */ - enum MessageState recv_state; - - /** - * Flags for the currently being received message from the PSYC service. - */ - enum GNUNET_PSYC_MessageFlags recv_flags; - - /** - * Expected value size for the modifier being received from the PSYC service. - */ - uint32_t recv_mod_value_size_expected; - - /** - * Actual value size for the modifier being received from the PSYC service. - */ - uint32_t recv_mod_value_size; - - /** - * Is transmission paused? - */ - uint8_t tmit_paused; - - /** - * Are we still waiting for a PSYC_TRANSMIT_ACK? - */ - uint8_t tmit_ack_pending; + struct GNUNET_MessageHeader *connect_msg; /** * Are we polling for incoming messages right now? @@ -177,11 +79,6 @@ struct GNUNET_PSYC_Channel uint8_t in_receive; /** - * Are we currently transmitting a message? - */ - uint8_t in_transmit; - - /** * Is this a master or slave channel? */ uint8_t is_master; @@ -193,7 +90,7 @@ struct GNUNET_PSYC_Channel */ struct GNUNET_PSYC_Master { - struct GNUNET_PSYC_Channel ch; + struct GNUNET_PSYC_Channel chn; GNUNET_PSYC_MasterStartCallback start_cb; @@ -201,6 +98,11 @@ struct GNUNET_PSYC_Master * Join request callback. */ GNUNET_PSYC_JoinRequestCallback join_req_cb; + + /** + * Closure for the callbacks. + */ + void *cb_cls; }; @@ -209,11 +111,16 @@ struct GNUNET_PSYC_Master */ struct GNUNET_PSYC_Slave { - struct GNUNET_PSYC_Channel ch; + struct GNUNET_PSYC_Channel chn; GNUNET_PSYC_SlaveConnectCallback connect_cb; GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb; + + /** + * Closure for the callbacks. + */ + void *cb_cls; }; @@ -258,934 +165,170 @@ struct GNUNET_PSYC_StateQuery static void -reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); - - -static void -channel_transmit_data (struct GNUNET_PSYC_Channel *ch); - - -/** - * Reschedule a connect attempt to the service. - * - * @param ch Channel to reconnect. - */ -static void -reschedule_connect (struct GNUNET_PSYC_Channel *ch) +channel_send_connect_msg (struct GNUNET_PSYC_Channel *chn) { - GNUNET_assert (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; - } - ch->in_receive = GNUNET_NO; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Scheduling task to reconnect to PSYC service in %s.\n", - GNUNET_STRINGS_relative_time_to_string (ch->reconnect_delay, GNUNET_YES)); - ch->reconnect_task = - GNUNET_SCHEDULER_add_delayed (ch->reconnect_delay, &reconnect, ch); - ch->reconnect_delay = GNUNET_TIME_STD_BACKOFF (ch->reconnect_delay); + uint16_t cmsg_size = ntohs (chn->connect_msg->size); + struct GNUNET_MessageHeader * cmsg = GNUNET_malloc (cmsg_size); + memcpy (cmsg, chn->connect_msg, cmsg_size); + GNUNET_CLIENT_MANAGER_transmit_now (chn->client, cmsg); } -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC channel handle - */ -static void -transmit_next (struct GNUNET_PSYC_Channel *ch); - - -/** - * Reset stored data related to the last received message. - */ static void -recv_reset (struct GNUNET_PSYC_Channel *ch) +channel_recv_disconnect (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - ch->recv_state = MSG_STATE_START; - ch->recv_flags = 0; - ch->recv_message_id = 0; - //FIXME: ch->recv_slave_key = { 0 }; - ch->recv_mod_value_size = 0; - ch->recv_mod_value_size_expected = 0; + struct GNUNET_PSYC_Channel * + chn = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*chn)); + GNUNET_CLIENT_MANAGER_reconnect (client); + channel_send_connect_msg (chn); } static void -recv_error (struct GNUNET_PSYC_Channel *ch) +channel_recv_message (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - GNUNET_PSYC_MessageCallback message_cb - = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC - ? ch->hist_message_cb - : ch->message_cb; - - if (NULL != message_cb) - message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, NULL); - - recv_reset (ch); + struct GNUNET_PSYC_Channel * + chn = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*chn)); + GNUNET_PSYC_receive_message (chn->recv, + (const struct GNUNET_PSYC_MessageHeader *) msg); } -/** - * Queue a message part for transmission to the PSYC service. - * - * The message part is added to the current message buffer. - * When this buffer is full, it is added to the transmission queue. - * - * @param ch Channel struct for the client. - * @param msg Modifier message part, or NULL when there's no more modifiers. - * @param end End of message. - */ static void -queue_message (struct GNUNET_PSYC_Channel *ch, - const struct GNUNET_MessageHeader *msg, - uint8_t end) +channel_recv_message_ack (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - uint16_t size = msg ? ntohs (msg->size) : 0; - - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Queueing message of type %u and size %u (end: %u)).\n", - ntohs (msg->type), size, end); - - struct MessageQueue *mq = ch->tmit_msg; - struct GNUNET_MessageHeader *qmsg = NULL; - if (NULL != mq) - { - qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - if (NULL == msg - || GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD < qmsg->size + size) - { - /* End of message or buffer is full, add it to transmission queue - * and start with empty buffer */ - qmsg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE); - qmsg->size = htons (qmsg->size); - GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq); - ch->tmit_msg = mq = NULL; - ch->tmit_ack_pending++; - } - else - { - /* Message fits in current buffer, append */ - ch->tmit_msg - = mq = GNUNET_realloc (mq, sizeof (*mq) + qmsg->size + size); - qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - memcpy ((char *) qmsg + qmsg->size, msg, size); - qmsg->size += size; - } - } - - if (NULL == mq && NULL != msg) - { - /* Empty buffer, copy over message. */ - ch->tmit_msg - = mq = GNUNET_malloc (sizeof (*mq) + sizeof (*qmsg) + size); - qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - qmsg->size = sizeof (*qmsg) + size; - memcpy (&qmsg[1], msg, size); - } - - if (NULL != mq - && (GNUNET_YES == end - || (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD - < qmsg->size + sizeof (struct GNUNET_MessageHeader)))) - { - /* End of message or buffer is full, add it to transmission queue. */ - qmsg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE); - qmsg->size = htons (qmsg->size); - GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq); - ch->tmit_msg = mq = NULL; - ch->tmit_ack_pending++; - } - - if (GNUNET_YES == end) - ch->in_transmit = GNUNET_NO; - - transmit_next (ch); + struct GNUNET_PSYC_Channel * + chn = GNUNET_CLIENT_MANAGER_get_user_context_ (client, sizeof (*chn)); + GNUNET_PSYC_transmit_got_ack (chn->tmit); } -/** - * Request a modifier from a client to transmit. - * - * @param mst Master handle. - */ static void -channel_transmit_mod (struct GNUNET_PSYC_Channel *ch) +master_recv_start_ack (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - uint16_t max_data_size, data_size; - char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = ""; - struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data; - int notify_ret; - - switch (ch->tmit.state) - { - case MSG_STATE_MODIFIER: - { - struct GNUNET_PSYC_MessageModifier *mod - = (struct GNUNET_PSYC_MessageModifier *) msg; - max_data_size = data_size = GNUNET_PSYC_MODIFIER_MAX_PAYLOAD; - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER); - msg->size = sizeof (struct GNUNET_PSYC_MessageModifier); - notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls, &data_size, &mod[1], - &mod->oper, &mod->value_size); - mod->name_size = strnlen ((char *) &mod[1], data_size); - if (mod->name_size < data_size) - { - mod->value_size = htonl (mod->value_size); - mod->name_size = htons (mod->name_size); - } - else if (0 < data_size) - { - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Got invalid modifier name.\n"); - notify_ret = GNUNET_SYSERR; - } - break; - } - case MSG_STATE_MOD_CONT: - { - max_data_size = data_size = GNUNET_PSYC_MOD_CONT_MAX_PAYLOAD; - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT); - msg->size = sizeof (struct GNUNET_MessageHeader); - notify_ret = ch->tmit.notify_mod (ch->tmit.notify_cls, - &data_size, &msg[1], NULL, NULL); - break; - } - default: - GNUNET_assert (0); - } - - switch (notify_ret) - { - case GNUNET_NO: - if (0 == data_size) - { /* Transmission paused, nothing to send. */ - ch->tmit_paused = GNUNET_YES; - return; - } - ch->tmit.state = MSG_STATE_MOD_CONT; - break; - - case GNUNET_YES: - if (0 == data_size) - { - /* End of modifiers. */ - ch->tmit.state = MSG_STATE_DATA; - if (0 == ch->tmit_ack_pending) - channel_transmit_data (ch); - - return; - } - ch->tmit.state = MSG_STATE_MODIFIER; - break; - - default: - LOG (GNUNET_ERROR_TYPE_ERROR, - "MasterTransmitNotifyModifier returned error " - "when requesting a modifier.\n"); - - ch->tmit.state = MSG_STATE_CANCEL; - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL); - msg->size = htons (sizeof (*msg)); - - queue_message (ch, msg, GNUNET_YES); - return; - } - - if (0 < data_size) - { - GNUNET_assert (data_size <= max_data_size); - msg->size = htons (msg->size + data_size); - queue_message (ch, msg, GNUNET_NO); - } - - channel_transmit_mod (ch); -} + struct GNUNET_PSYC_Master * + mst = GNUNET_CLIENT_MANAGER_get_user_context_ (client, + sizeof (struct GNUNET_PSYC_Channel)); - -/** - * Request data from a client to transmit. - * - * @param mst Master handle. - */ -static void -channel_transmit_data (struct GNUNET_PSYC_Channel *ch) -{ - uint16_t data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD; - char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = ""; - struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data; - - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA); - - int notify_ret = ch->tmit.notify_data (ch->tmit.notify_cls, - &data_size, &msg[1]); - switch (notify_ret) - { - case GNUNET_NO: - if (0 == data_size) - { - /* Transmission paused, nothing to send. */ - ch->tmit_paused = GNUNET_YES; - return; - } - break; - - case GNUNET_YES: - ch->tmit.state = MSG_STATE_END; - break; - - default: - LOG (GNUNET_ERROR_TYPE_ERROR, - "MasterTransmitNotify returned error when requesting data.\n"); - - ch->tmit.state = MSG_STATE_CANCEL; - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL); - msg->size = htons (sizeof (*msg)); - queue_message (ch, msg, GNUNET_YES); - return; - } - - if (0 < data_size) - { - GNUNET_assert (data_size <= GNUNET_PSYC_DATA_MAX_PAYLOAD); - msg->size = htons (sizeof (*msg) + data_size); - queue_message (ch, msg, !notify_ret); - } - - /* End of message. */ - if (GNUNET_YES == notify_ret) - { - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END); - msg->size = htons (sizeof (*msg)); - queue_message (ch, msg, GNUNET_YES); - } + struct CountersResult *cres = (struct CountersResult *) msg; + if (NULL != mst->start_cb) + mst->start_cb (mst->cb_cls, GNUNET_ntohll (cres->max_message_id)); } -/** - * Send a message to a channel. - * - * @param ch Handle to the PSYC channel. - * @param method_name Which method should be invoked. - * @param notify_mod Function to call to obtain modifiers. - * @param notify_data Function to call to obtain fragments of the data. - * @param notify_cls Closure for @a notify_mod and @a notify_data. - * @param flags Flags for the message being transmitted. - * - * @return Transmission handle, NULL on error (i.e. more than one request queued). - */ -static struct GNUNET_PSYC_ChannelTransmitHandle * -channel_transmit (struct GNUNET_PSYC_Channel *ch, - const char *method_name, - GNUNET_PSYC_TransmitNotifyModifier notify_mod, - GNUNET_PSYC_TransmitNotifyData notify_data, - void *notify_cls, - uint32_t flags) -{ - if (GNUNET_NO != ch->in_transmit) - return NULL; - ch->in_transmit = GNUNET_YES; - - size_t size = strlen (method_name) + 1; - struct GNUNET_PSYC_MessageMethod *pmeth; - struct GNUNET_MessageHeader *qmsg; - struct MessageQueue * - mq = ch->tmit_msg = GNUNET_malloc (sizeof (*mq) + sizeof (*qmsg) - + sizeof (*pmeth) + size); - qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - qmsg->size = sizeof (*qmsg) + sizeof (*pmeth) + size; - - pmeth = (struct GNUNET_PSYC_MessageMethod *) &qmsg[1]; - pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD); - pmeth->header.size = htons (sizeof (*pmeth) + size); - pmeth->flags = htonl (flags); - memcpy (&pmeth[1], method_name, size); - - ch->tmit.ch = ch; - ch->tmit.notify_mod = notify_mod; - ch->tmit.notify_data = notify_data; - ch->tmit.notify_cls = notify_cls; - ch->tmit.state = MSG_STATE_MODIFIER; - - channel_transmit_mod (ch); - return &ch->tmit; -} - - -/** - * Resume transmission to the channel. - * - * @param th Handle of the request that is being resumed. - */ static void -channel_transmit_resume (struct GNUNET_PSYC_ChannelTransmitHandle *th) +master_recv_join_request (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - struct GNUNET_PSYC_Channel *ch = th->ch; - if (0 == ch->tmit_ack_pending) - { - ch->tmit_paused = GNUNET_NO; - channel_transmit_data (ch); - } -} + struct GNUNET_PSYC_Master * + mst = GNUNET_CLIENT_MANAGER_get_user_context_ (client, + sizeof (struct GNUNET_PSYC_Channel)); + const struct MasterJoinRequest *req = (const struct MasterJoinRequest *) msg; -/** - * Abort transmission request to channel. - * - * @param th Handle of the request that is being aborted. - */ -static void -channel_transmit_cancel (struct GNUNET_PSYC_ChannelTransmitHandle *th) -{ - struct GNUNET_PSYC_Channel *ch = th->ch; - if (GNUNET_NO == ch->in_transmit) - return; -} + struct GNUNET_PSYC_MessageHeader *pmsg = NULL; + if (ntohs (req->header.size) <= sizeof (*req) + sizeof (*pmsg)) + pmsg = (struct GNUNET_PSYC_MessageHeader *) &req[1]; + struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); + jh->mst = mst; + jh->slave_key = req->slave_key; -/** - * Handle incoming message from the PSYC service. - * - * @param ch The channel the message is sent to. - * @param pmsg The message. - */ -static void -handle_psyc_message (struct GNUNET_PSYC_Channel *ch, - const struct GNUNET_PSYC_MessageHeader *msg) -{ - uint16_t size = ntohs (msg->header.size); - uint32_t flags = ntohl (msg->flags); - - GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, - (struct GNUNET_MessageHeader *) msg); - - if (MSG_STATE_START == ch->recv_state) - { - ch->recv_message_id = GNUNET_ntohll (msg->message_id); - ch->recv_flags = flags; - ch->recv_slave_key = msg->slave_key; - ch->recv_mod_value_size = 0; - ch->recv_mod_value_size_expected = 0; - } - else if (GNUNET_ntohll (msg->message_id) != ch->recv_message_id) - { - // FIXME - LOG (GNUNET_ERROR_TYPE_WARNING, - "Unexpected message ID. Got: %" PRIu64 ", expected: %" PRIu64 "\n", - GNUNET_ntohll (msg->message_id), ch->recv_message_id); - GNUNET_break_op (0); - recv_error (ch); - return; - } - else if (flags != ch->recv_flags) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Unexpected message flags. Got: %lu, expected: %lu\n", - flags, ch->recv_flags); - GNUNET_break_op (0); - recv_error (ch); - return; - } - - uint16_t pos = 0, psize = 0, ptype, size_eq, size_min; - - for (pos = 0; sizeof (*msg) + pos < size; pos += psize) - { - const struct GNUNET_MessageHeader *pmsg - = (const struct GNUNET_MessageHeader *) ((char *) &msg[1] + pos); - psize = ntohs (pmsg->size); - ptype = ntohs (pmsg->type); - size_eq = size_min = 0; - - if (psize < sizeof (*pmsg) || sizeof (*msg) + pos + psize > size) - { - GNUNET_log (GNUNET_ERROR_TYPE_WARNING, - "Dropping message of type %u with invalid size %u.\n", - ptype, psize); - recv_error (ch); - return; - } - - GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, - "Received message part from PSYC.\n"); - GNUNET_PSYC_log_message (GNUNET_ERROR_TYPE_DEBUG, pmsg); - - switch (ptype) - { - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD: - size_min = sizeof (struct GNUNET_PSYC_MessageMethod); - break; - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER: - size_min = sizeof (struct GNUNET_PSYC_MessageModifier); - break; - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT: - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA: - size_min = sizeof (struct GNUNET_MessageHeader); - break; - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END: - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL: - size_eq = sizeof (struct GNUNET_MessageHeader); - break; - default: - GNUNET_break_op (0); - recv_error (ch); - return; - } - - if (! ((0 < size_eq && psize == size_eq) - || (0 < size_min && size_min <= psize))) - { - GNUNET_break_op (0); - recv_error (ch); - return; - } - - switch (ptype) - { - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD: - { - struct GNUNET_PSYC_MessageMethod *meth - = (struct GNUNET_PSYC_MessageMethod *) pmsg; - - if (MSG_STATE_START != ch->recv_state) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Dropping out of order message method (%u).\n", - ch->recv_state); - /* It is normal to receive an incomplete message right after connecting, - * but should not happen later. - * FIXME: add a check for this condition. - */ - GNUNET_break_op (0); - recv_error (ch); - return; - } - - if ('\0' != *((char *) meth + psize - 1)) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Dropping message with malformed method. " - "Message ID: %" PRIu64 "\n", ch->recv_message_id); - GNUNET_break_op (0); - recv_error (ch); - return; - } - ch->recv_state = MSG_STATE_METHOD; - break; - } - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER: - { - if (!(MSG_STATE_METHOD == ch->recv_state - || MSG_STATE_MODIFIER == ch->recv_state - || MSG_STATE_MOD_CONT == ch->recv_state)) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Dropping out of order message modifier (%u).\n", - ch->recv_state); - GNUNET_break_op (0); - recv_error (ch); - return; - } - - struct GNUNET_PSYC_MessageModifier *mod - = (struct GNUNET_PSYC_MessageModifier *) pmsg; - - uint16_t name_size = ntohs (mod->name_size); - ch->recv_mod_value_size_expected = ntohl (mod->value_size); - ch->recv_mod_value_size = psize - sizeof (*mod) - name_size - 1; - - if (psize < sizeof (*mod) + name_size + 1 - || '\0' != *((char *) &mod[1] + name_size) - || ch->recv_mod_value_size_expected < ch->recv_mod_value_size) - { - LOG (GNUNET_ERROR_TYPE_WARNING, "Dropping malformed modifier.\n"); - GNUNET_break_op (0); - recv_error (ch); - return; - } - ch->recv_state = MSG_STATE_MODIFIER; - break; - } - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT: - { - ch->recv_mod_value_size += psize - sizeof (*pmsg); - - if (!(MSG_STATE_MODIFIER == ch->recv_state - || MSG_STATE_MOD_CONT == ch->recv_state) - || ch->recv_mod_value_size_expected < ch->recv_mod_value_size) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Dropping out of order message modifier continuation " - "!(%u == %u || %u == %u) || %lu < %lu.\n", - MSG_STATE_MODIFIER, ch->recv_state, - MSG_STATE_MOD_CONT, ch->recv_state, - ch->recv_mod_value_size_expected, ch->recv_mod_value_size); - GNUNET_break_op (0); - recv_error (ch); - return; - } - break; - } - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA: - { - if (ch->recv_state < MSG_STATE_METHOD - || ch->recv_mod_value_size_expected != ch->recv_mod_value_size) - { - LOG (GNUNET_ERROR_TYPE_WARNING, - "Dropping out of order message data fragment " - "(%u < %u || %lu != %lu).\n", - ch->recv_state, MSG_STATE_METHOD, - ch->recv_mod_value_size_expected, ch->recv_mod_value_size); - - GNUNET_break_op (0); - recv_error (ch); - return; - } - ch->recv_state = MSG_STATE_DATA; - break; - } - } - - GNUNET_PSYC_MessageCallback message_cb - = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC - ? ch->hist_message_cb - : ch->message_cb; - - if (NULL != message_cb) - message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, pmsg); - - switch (ptype) - { - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_END: - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_CANCEL: - recv_reset (ch); - break; - } - } + if (NULL != mst->join_req_cb) + mst->join_req_cb (mst->cb_cls, &req->slave_key, pmsg, jh); } -/** - * Handle incoming message acknowledgement from the PSYC service. - * - * @param ch The channel the acknowledgement is sent to. - */ static void -handle_psyc_message_ack (struct GNUNET_PSYC_Channel *ch) +slave_recv_join_ack (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - if (0 == ch->tmit_ack_pending) - { - LOG (GNUNET_ERROR_TYPE_WARNING, "Ignoring extraneous message ACK\n"); - GNUNET_break (0); - return; - } - ch->tmit_ack_pending--; - - switch (ch->tmit.state) - { - case MSG_STATE_MODIFIER: - case MSG_STATE_MOD_CONT: - if (GNUNET_NO == ch->tmit_paused) - channel_transmit_mod (ch); - break; - - case MSG_STATE_DATA: - if (GNUNET_NO == ch->tmit_paused) - channel_transmit_data (ch); - break; - - case MSG_STATE_END: - case MSG_STATE_CANCEL: - break; - - default: - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Ignoring message ACK in state %u.\n", ch->tmit.state); - } + struct GNUNET_PSYC_Slave * + slv = GNUNET_CLIENT_MANAGER_get_user_context_ (client, + sizeof (struct GNUNET_PSYC_Channel)); + struct CountersResult *cres = (struct CountersResult *) msg; + if (NULL != slv->connect_cb) + slv->connect_cb (slv->cb_cls, GNUNET_ntohll (cres->max_message_id)); } static void -handle_psyc_join_request (struct GNUNET_PSYC_Master *mst, - const struct MasterJoinRequest *req) +slave_recv_join_decision (void *cls, + struct GNUNET_CLIENT_MANAGER_Connection *client, + const struct GNUNET_MessageHeader *msg) { - struct GNUNET_PSYC_MessageHeader *msg = NULL; - if (ntohs (req->header.size) <= sizeof (*req) + sizeof (*msg)) - msg = (struct GNUNET_PSYC_MessageHeader *) &req[1]; + struct GNUNET_PSYC_Slave * + slv = GNUNET_CLIENT_MANAGER_get_user_context_ (client, + sizeof (struct GNUNET_PSYC_Channel)); + const struct SlaveJoinDecision * + dcsn = (const struct SlaveJoinDecision *) msg; - struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); - jh->mst = mst; - jh->slave_key = req->slave_key; + struct GNUNET_PSYC_MessageHeader *pmsg = NULL; + if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg)) + pmsg = (struct GNUNET_PSYC_MessageHeader *) &dcsn[1]; - if (NULL != mst->join_req_cb) - mst->join_req_cb (mst->ch.cb_cls, &req->slave_key, msg, jh); + struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); + if (NULL != slv->join_dcsn_cb) + slv->join_dcsn_cb (slv->cb_cls, ntohl (dcsn->is_admitted), pmsg); } -static void -handle_psyc_join_decision (struct GNUNET_PSYC_Slave *slv, - const struct SlaveJoinDecision *dcsn) +static struct GNUNET_CLIENT_MANAGER_MessageHandler master_handlers[] = { - struct GNUNET_PSYC_MessageHeader *msg = NULL; - if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*msg)) - msg = (struct GNUNET_PSYC_MessageHeader *) &dcsn[1]; + { &channel_recv_disconnect, NULL, 0, 0, GNUNET_NO }, - struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); - if (NULL != slv->join_dcsn_cb) - slv->join_dcsn_cb (slv->ch.cb_cls, ntohl (dcsn->is_admitted), msg); -} + { &channel_recv_message, NULL, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, + sizeof (struct GNUNET_PSYC_MessageHeader), GNUNET_YES }, + { &channel_recv_message_ack, NULL, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK, + sizeof (struct GNUNET_MessageHeader), GNUNET_NO }, -/** - * 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) - { - // timeout / disconnected from service, reconnect - reschedule_connect (ch); - return; - } - uint16_t size_eq = 0; - uint16_t size_min = 0; - uint16_t size = ntohs (msg->size); - uint16_t type = ntohs (msg->type); - - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Received message of type %d and size %u from PSYC service\n", - type, size); - - switch (type) - { - case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK: - case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK: - size_eq = sizeof (struct CountersResult); - break; - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE: - size_min = sizeof (struct GNUNET_PSYC_MessageHeader); - break; - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK: - size_eq = sizeof (struct GNUNET_MessageHeader); - break; - case GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST: - size_min = sizeof (struct MasterJoinRequest); - break; - case GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION: - size_min = sizeof (struct SlaveJoinDecision); - break; - default: - GNUNET_break_op (0); - return; - } - - if (! ((0 < size_eq && size == size_eq) - || (0 < size_min && size_min <= size))) - { - GNUNET_break_op (0); - return; - } - - switch (type) - { - case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK: - { - struct CountersResult *cres = (struct CountersResult *) msg; - if (NULL != mst->start_cb) - mst->start_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id)); - break; - } - case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK: - { - struct CountersResult *cres = (struct CountersResult *) msg; - if (NULL != slv->connect_cb) - slv->connect_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id)); - break; - } - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK: - { - handle_psyc_message_ack (ch); - break; - } - - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE: - handle_psyc_message (ch, (const struct GNUNET_PSYC_MessageHeader *) msg); - break; - - case GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST: - handle_psyc_join_request ((struct GNUNET_PSYC_Master *) ch, - (const struct MasterJoinRequest *) msg); - break; - - case GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION: - handle_psyc_join_decision ((struct GNUNET_PSYC_Slave *) ch, - (const struct SlaveJoinDecision *) msg); - break; - } - - if (NULL != ch->client) - { - GNUNET_CLIENT_receive (ch->client, &message_handler, ch, - GNUNET_TIME_UNIT_FOREVER_REL); - } -} + { &master_recv_start_ack, NULL, + GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK, + sizeof (struct CountersResult), GNUNET_NO }, + { &master_recv_join_request, NULL, + GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST, + sizeof (struct MasterJoinRequest), GNUNET_YES }, -/** - * Transmit next message to service. - * - * @param cls The struct GNUNET_PSYC_Channel. - * @param size Number of bytes available in @a buf. - * @param buf Where to copy the message. - * - * @return Number of bytes copied to @a buf. - */ -static size_t -send_next_message (void *cls, size_t size, void *buf) -{ - LOG (GNUNET_ERROR_TYPE_DEBUG, "send_next_message()\n"); - struct GNUNET_PSYC_Channel *ch = cls; - struct MessageQueue *mq = ch->tmit_head; - if (NULL == mq) - return 0; - struct GNUNET_MessageHeader *qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - size_t ret = ntohs (qmsg->size); - ch->th = NULL; - if (ret > size) - { - reschedule_connect (ch); - return 0; - } - memcpy (buf, qmsg, ret); - - GNUNET_CONTAINER_DLL_remove (ch->tmit_head, ch->tmit_tail, mq); - GNUNET_free (mq); - - if (NULL != ch->tmit_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; -} + { NULL, NULL, 0, 0, GNUNET_NO } +}; -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC handle. - */ -static void -transmit_next (struct GNUNET_PSYC_Channel *ch) +static struct GNUNET_CLIENT_MANAGER_MessageHandler slave_handlers[] = { - LOG (GNUNET_ERROR_TYPE_DEBUG, "transmit_next()\n"); - if (NULL != ch->th || NULL == ch->client) - return; - - struct MessageQueue *mq = ch->tmit_head; - if (NULL == mq) - return; - struct GNUNET_MessageHeader *qmsg = (struct GNUNET_MessageHeader *) &mq[1]; - - ch->th = GNUNET_CLIENT_notify_transmit_ready (ch->client, - ntohs (qmsg->size), - GNUNET_TIME_UNIT_FOREVER_REL, - GNUNET_NO, - &send_next_message, - ch); -} + { &channel_recv_disconnect, NULL, 0, 0, GNUNET_NO }, + { &channel_recv_message, NULL, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, + sizeof (struct GNUNET_PSYC_MessageHeader), GNUNET_YES }, -/** - * 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; - - recv_reset (ch); - 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); - uint16_t reconn_size = ntohs (ch->reconnect_msg->size); - - if (NULL == ch->tmit_head || - 0 != memcmp (&ch->tmit_head[1], ch->reconnect_msg, reconn_size)) - { - struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + reconn_size); - memcpy (&mq[1], ch->reconnect_msg, reconn_size); - GNUNET_CONTAINER_DLL_insert (ch->tmit_head, ch->tmit_tail, mq); - } - transmit_next (ch); -} + { &channel_recv_message_ack, NULL, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK, + sizeof (struct GNUNET_MessageHeader), GNUNET_NO }, + { &slave_recv_join_ack, NULL, + GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK, + sizeof (struct CountersResult), GNUNET_NO }, -/** - * Disconnect from the PSYC service. - * - * @param c Channel handle to disconnect. - */ -static void -disconnect (void *c) -{ - struct GNUNET_PSYC_Channel *ch = c; - - GNUNET_assert (NULL != ch); - if (ch->tmit_head != ch->tmit_tail) - { - LOG (GNUNET_ERROR_TYPE_ERROR, - "Disconnecting while there are still outstanding messages!\n"); - GNUNET_break (0); - } - 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) - { - GNUNET_free (ch->reconnect_msg); - ch->reconnect_msg = NULL; - } -} + { &slave_recv_join_decision, NULL, + GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION, + sizeof (struct SlaveJoinDecision), GNUNET_YES }, + + { NULL, NULL, 0, 0, GNUNET_NO } +}; /** @@ -1227,24 +370,29 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls) { struct GNUNET_PSYC_Master *mst = GNUNET_malloc (sizeof (*mst)); - struct GNUNET_PSYC_Channel *ch = &mst->ch; - struct MasterStartRequest *req = GNUNET_malloc (sizeof (*req)); + struct GNUNET_PSYC_Channel *chn = &mst->chn; + struct MasterStartRequest *req = GNUNET_malloc (sizeof (*req)); req->header.size = htons (sizeof (*req)); req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START); req->channel_key = *channel_key; req->policy = policy; + chn->connect_msg = (struct GNUNET_MessageHeader *) req; + chn->cfg = cfg; + chn->is_master = GNUNET_YES; + mst->start_cb = start_cb; mst->join_req_cb = join_request_cb; - ch->message_cb = message_cb; - ch->cb_cls = cls; - ch->cfg = cfg; - ch->is_master = GNUNET_YES; - ch->reconnect_msg = (struct GNUNET_MessageHeader *) req; - ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO; - ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, mst); + mst->cb_cls = cls; + + chn->client = GNUNET_CLIENT_MANAGER_connect (cfg, "psyc", master_handlers); + GNUNET_CLIENT_MANAGER_set_user_context_ (chn->client, mst, sizeof (*chn)); + + chn->tmit = GNUNET_PSYC_transmit_create (chn->client); + chn->recv = GNUNET_PSYC_receive_create (message_cb, message_cb, cls); + channel_send_connect_msg (chn); return mst; } @@ -1253,12 +401,13 @@ GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg, * Stop a PSYC master channel. * * @param master PSYC channel master to stop. + * @param keep_active FIXME */ void -GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master) +GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *mst) { - disconnect (master); - GNUNET_free (master); + GNUNET_CLIENT_MANAGER_disconnect (mst->chn.client, GNUNET_YES); + GNUNET_free (mst); } @@ -1292,7 +441,7 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh, const struct GNUNET_PeerIdentity *relays, const struct GNUNET_PSYC_MessageHeader *join_resp) { - struct GNUNET_PSYC_Channel *ch = &jh->mst->ch; + struct GNUNET_PSYC_Channel *chn = &jh->mst->chn; struct MasterJoinDecision *dcsn; uint16_t join_resp_size = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0; @@ -1302,9 +451,7 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh, < sizeof (*dcsn) + relay_size + join_resp_size) return GNUNET_SYSERR; - struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*dcsn) - + relay_size + join_resp_size); - dcsn = (struct MasterJoinDecision *) &mq[1]; + dcsn = GNUNET_malloc (sizeof (*dcsn) + relay_size + join_resp_size); dcsn->header.size = htons (sizeof (*dcsn) + relay_size + join_resp_size); dcsn->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION); dcsn->is_admitted = htonl (is_admitted); @@ -1313,8 +460,7 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh, if (0 < join_resp_size) memcpy (&dcsn[1], join_resp, join_resp_size); - GNUNET_CONTAINER_DLL_insert_tail (ch->tmit_head, ch->tmit_tail, mq); - transmit_next (ch); + GNUNET_CLIENT_MANAGER_transmit (chn->client, &dcsn->header); return GNUNET_OK; } @@ -1332,40 +478,59 @@ GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh, * @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, GNUNET_PSYC_TransmitNotifyModifier notify_mod, GNUNET_PSYC_TransmitNotifyData notify_data, void *notify_cls, enum GNUNET_PSYC_MasterTransmitFlags flags) { - return (struct GNUNET_PSYC_MasterTransmitHandle *) - channel_transmit (&master->ch, method_name, notify_mod, notify_data, - notify_cls, flags); + if (GNUNET_OK + == GNUNET_PSYC_transmit_message (mst->chn.tmit, method_name, NULL, + notify_mod, notify_data, notify_cls, + flags)) + return (struct GNUNET_PSYC_MasterTransmitHandle *) mst->chn.tmit; + else + return NULL; } /** * Resume transmission to the channel. * - * @param th Handle of the request that is being resumed. + * @param tmit Handle of the request that is being resumed. */ void -GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *th) +GNUNET_PSYC_master_transmit_resume (struct GNUNET_PSYC_MasterTransmitHandle *tmit) { - channel_transmit_resume ((struct GNUNET_PSYC_ChannelTransmitHandle *) th); + GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit); } /** * Abort transmission request to the channel. * - * @param th Handle of the request that is being aborted. + * @param tmit Handle of the request that is being aborted. */ void -GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th) +GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *tmit) { - channel_transmit_cancel ((struct GNUNET_PSYC_ChannelTransmitHandle *) th); + GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit); +} + + +/** + * 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. + */ +struct GNUNET_PSYC_Channel * +GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master) +{ + return &master->chn; } @@ -1420,7 +585,7 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg, uint16_t data_size) { struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv)); - struct GNUNET_PSYC_Channel *ch = &slv->ch; + struct GNUNET_PSYC_Channel *chn = &slv->chn; struct SlaveJoinRequest *req = GNUNET_malloc (sizeof (*req) + relay_count * sizeof (*relays)); req->header.size = htons (sizeof (*req) @@ -1432,17 +597,21 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg, req->relay_count = htonl (relay_count); memcpy (&req[1], relays, relay_count * sizeof (*relays)); + chn->connect_msg = (struct GNUNET_MessageHeader *) req; + chn->cfg = cfg; + chn->is_master = GNUNET_NO; + slv->connect_cb = connect_cb; slv->join_dcsn_cb = join_decision_cb; - ch->message_cb = message_cb; - ch->cb_cls = cls; + slv->cb_cls = cls; - ch->cfg = cfg; - ch->is_master = GNUNET_NO; - ch->reconnect_msg = (struct GNUNET_MessageHeader *) req; - ch->reconnect_delay = GNUNET_TIME_UNIT_ZERO; - ch->reconnect_task = GNUNET_SCHEDULER_add_now (&reconnect, slv); + chn->client = GNUNET_CLIENT_MANAGER_connect (cfg, "psyc", slave_handlers); + GNUNET_CLIENT_MANAGER_set_user_context_ (chn->client, slv, sizeof (*chn)); + chn->recv = GNUNET_PSYC_receive_create (message_cb, message_cb, cls); + chn->tmit = GNUNET_PSYC_transmit_create (chn->client); + + channel_send_connect_msg (chn); return slv; } @@ -1456,10 +625,10 @@ GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg, * @param slave Slave handle. */ void -GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave) +GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slv) { - disconnect (slave); - GNUNET_free (slave); + GNUNET_CLIENT_MANAGER_disconnect (slv->chn.client, GNUNET_YES); + GNUNET_free (slv); } @@ -1477,69 +646,59 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave) * queued). */ struct GNUNET_PSYC_SlaveTransmitHandle * -GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave, +GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slv, const char *method_name, GNUNET_PSYC_TransmitNotifyModifier notify_mod, GNUNET_PSYC_TransmitNotifyData notify_data, void *notify_cls, enum GNUNET_PSYC_SlaveTransmitFlags flags) + { - return (struct GNUNET_PSYC_SlaveTransmitHandle *) - channel_transmit (&slave->ch, method_name, - notify_mod, notify_data, notify_cls, flags); + if (GNUNET_OK + == GNUNET_PSYC_transmit_message (slv->chn.tmit, method_name, NULL, + notify_mod, notify_data, notify_cls, + flags)) + return (struct GNUNET_PSYC_SlaveTransmitHandle *) slv->chn.tmit; + else + return NULL; } /** * Resume transmission to the master. * - * @param th Handle of the request that is being resumed. + * @param tmit Handle of the request that is being resumed. */ void -GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *th) +GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *tmit) { - channel_transmit_resume ((struct GNUNET_PSYC_ChannelTransmitHandle *) th); + GNUNET_PSYC_transmit_resume ((struct GNUNET_PSYC_TransmitHandle *) tmit); } /** * Abort transmission request to master. * - * @param th Handle of the request that is being aborted. + * @param tmit Handle of the request that is being aborted. */ void -GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *th) +GNUNET_PSYC_slave_transmit_cancel (struct GNUNET_PSYC_SlaveTransmitHandle *tmit) { - channel_transmit_cancel ((struct GNUNET_PSYC_ChannelTransmitHandle *) th); -} - - -/** - * 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. - */ -struct GNUNET_PSYC_Channel * -GNUNET_PSYC_master_get_channel (struct GNUNET_PSYC_Master *master) -{ - return &master->ch; + GNUNET_PSYC_transmit_cancel ((struct GNUNET_PSYC_TransmitHandle *) tmit); } /** * Convert @a slave to a @e channel handle to access the @e channel APIs. * - * @param slave Slave handle. + * @param slv Slave handle. * * @return Channel handle, valid for as long as @a slave is valid. */ struct GNUNET_PSYC_Channel * -GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slave) +GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slv) { - return &slave->ch; + return &slv->chn; } @@ -1565,23 +724,17 @@ GNUNET_PSYC_slave_get_channel (struct GNUNET_PSYC_Slave *slave) * @param effective_since Addition of slave is in effect since this message ID. */ void -GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel, +GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key, uint64_t announced_at, uint64_t effective_since) { - struct ChannelSlaveAdd *slvadd; - struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*slvadd)); - - slvadd = (struct ChannelSlaveAdd *) &mq[1]; - slvadd->header.type = htons (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_tail (channel->tmit_head, - channel->tmit_tail, - mq); - transmit_next (channel); + struct ChannelSlaveAdd *add = GNUNET_malloc (sizeof (*add)); + add->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD); + add->header.size = htons (sizeof (*add)); + add->announced_at = GNUNET_htonll (announced_at); + add->effective_since = GNUNET_htonll (effective_since); + GNUNET_CLIENT_MANAGER_transmit (chn->client, &add->header); } @@ -1607,21 +760,15 @@ GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel, * @param announced_at ID of the message that announced the membership change. */ void -GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel, +GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn, const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key, uint64_t announced_at) { - struct ChannelSlaveRemove *slvrm; - struct MessageQueue *mq = GNUNET_malloc (sizeof (*mq) + sizeof (*slvrm)); - - slvrm = (struct ChannelSlaveRemove *) &mq[1]; - slvrm->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM); - slvrm->header.size = htons (sizeof (*slvrm)); - slvrm->announced_at = GNUNET_htonll (announced_at); - GNUNET_CONTAINER_DLL_insert_tail (channel->tmit_head, - channel->tmit_tail, - mq); - transmit_next (channel); + struct ChannelSlaveRemove *rm = GNUNET_malloc (sizeof (*rm)); + rm->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM); + rm->header.size = htons (sizeof (*rm)); + rm->announced_at = GNUNET_htonll (announced_at); + GNUNET_CLIENT_MANAGER_transmit (chn->client, &rm->header); } |