X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fpsyc%2Fpsyc_api.c;h=9c58322934fa1abfc9d953b6493fdac3e1b171ad;hb=7afd744770424ea38ceb8aa41b9eb94f5c089f87;hp=85f86ceaa203fe3c2b78d4d0a0098560227a55b8;hpb=093f0291be26fa3dfc6fc98a536028ef99517832;p=oweals%2Fgnunet.git diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c index 85f86ceaa..9c5832293 100644 --- a/src/psyc/psyc_api.c +++ b/src/psyc/psyc_api.c @@ -1,6 +1,6 @@ /* * This file is part of GNUnet - * (C) 2013 Christian Grothoff (and other contributing authors) + * Copyright (C) 2013 GNUnet e.V. * * GNUnet is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published @@ -14,8 +14,8 @@ * * You should have received a copy of the GNU General Public License * along with GNUnet; see the file COPYING. If not, write to the - * Free Software Foundation, Inc., 59 Temple Place - Suite 330, - * Boston, MA 02111-1307, USA. + * Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor, + * Boston, MA 02110-1301, USA. */ /** @@ -34,146 +34,68 @@ #include "platform.h" #include "gnunet_util_lib.h" -#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; -}; - - -/** - * 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. */ struct GNUNET_PSYC_Channel { - /** - * Transmission handle; - */ - struct GNUNET_PSYC_ChannelTransmitHandle tmit; - /** * Configuration to use. */ const struct GNUNET_CONFIGURATION_Handle *cfg; /** - * Socket (if available). + * Client connection to the service. */ - struct GNUNET_CLIENT_Connection *client; + struct GNUNET_MQ_Handle *mq; /** - * Currently pending transmission request, or NULL for none. + * Message to send on connect. */ - struct GNUNET_CLIENT_TransmitHandle *th; + struct GNUNET_MQ_Envelope *connect_env; /** - * Head of messages to transmit to the service. - */ - struct MessageQueue *tmit_head; - - /** - * Tail of operations to transmit to the service. - */ - struct MessageQueue *tmit_tail; - - /** - * Message currently being transmitted to the service. - */ - struct MessageQueue *tmit_msg; - - /** - * 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. + * Time to wait until we try to reconnect on failure. */ 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; - - /** - * Join handler callback. - */ - GNUNET_PSYC_JoinCallback join_cb; - - /** - * Closure for @a message_cb and @a join_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. + * Task for reconnecting when the listener fails. */ - struct GNUNET_CRYPTO_EddsaPublicKey recv_slave_key; + struct GNUNET_SCHEDULER_Task *reconnect_task; /** - * State of the currently being received message from the PSYC service. + * Async operations. */ - enum MessageState recv_state; + struct GNUNET_OP_Handle *op; /** - * 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. + * Transmission handle; */ - uint32_t recv_mod_value_size_expected; + struct GNUNET_PSYC_TransmitHandle *tmit; /** - * Actual value size for the modifier being received from the PSYC service. + * Receipt handle; */ - uint32_t recv_mod_value_size; + struct GNUNET_PSYC_ReceiveHandle *recv; /** - * Is transmission paused? + * Function called after disconnected from the service. */ - uint8_t tmit_paused; + GNUNET_ContinuationCallback disconnect_cb; /** - * Are we still waiting for a PSYC_TRANSMIT_ACK? + * Closure for @a disconnect_cb. */ - uint8_t tmit_ack_pending; + void *disconnect_cls; /** * Are we polling for incoming messages right now? @@ -181,14 +103,15 @@ struct GNUNET_PSYC_Channel uint8_t in_receive; /** - * Are we currently transmitting a message? + * Is this a master or slave channel? */ - uint8_t in_transmit; + uint8_t is_master; /** - * Is this a master or slave channel? + * Is this channel in the process of disconnecting from the service? + * #GNUNET_YES or #GNUNET_NO */ - uint8_t is_master; + uint8_t is_disconnecting; }; @@ -197,9 +120,19 @@ struct GNUNET_PSYC_Channel */ struct GNUNET_PSYC_Master { - struct GNUNET_PSYC_Channel ch; + struct GNUNET_PSYC_Channel chn; GNUNET_PSYC_MasterStartCallback start_cb; + + /** + * Join request callback. + */ + GNUNET_PSYC_JoinRequestCallback join_req_cb; + + /** + * Closure for the callbacks. + */ + void *cb_cls; }; @@ -208,21 +141,29 @@ 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_SlaveJoinCallback join_cb; + GNUNET_PSYC_JoinDecisionCallback join_dcsn_cb; + + /** + * Closure for the callbacks. + */ + void *cb_cls; }; /** * Handle that identifies a join request. * - * Used to match calls to #GNUNET_PSYC_JoinCallback to the + * Used to match calls to #GNUNET_PSYC_JoinRequestCallback to the * corresponding calls to GNUNET_PSYC_join_decision(). */ struct GNUNET_PSYC_JoinHandle { - + struct GNUNET_PSYC_Master *mst; + struct GNUNET_CRYPTO_EcdsaPublicKey slave_pub_key; }; @@ -235,906 +176,523 @@ struct GNUNET_PSYC_SlaveTransmitHandle }; -/** - * Handle to a story telling operation. - */ -struct GNUNET_PSYC_Story +struct GNUNET_PSYC_HistoryRequest { + /** + * Channel. + */ + struct GNUNET_PSYC_Channel *chn; -}; + /** + * Operation ID. + */ + uint64_t op_id; + /** + * Message handler. + */ + struct GNUNET_PSYC_ReceiveHandle *recv; -/** - * Handle for a state query operation. - */ -struct GNUNET_PSYC_StateQuery -{ + /** + * Function to call when the operation finished. + */ + GNUNET_ResultCallback result_cb; + /** + * Closure for @a result_cb. + */ + void *cls; }; -static void -reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); +struct GNUNET_PSYC_StateRequest +{ + /** + * Channel. + */ + struct GNUNET_PSYC_Channel *chn; + + /** + * Operation ID. + */ + uint64_t op_id; + /** + * State variable result callback. + */ + GNUNET_PSYC_StateVarCallback var_cb; -static void -channel_transmit_data (struct GNUNET_PSYC_Channel *ch); + /** + * Function to call when the operation finished. + */ + GNUNET_ResultCallback result_cb; + /** + * Closure for @a result_cb. + */ + void *cls; +}; -/** - * Reschedule a connect attempt to the service. - * - * @param ch Channel to reconnect. - */ -static void -reschedule_connect (struct GNUNET_PSYC_Channel *ch) -{ - 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); +static int +check_channel_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + return GNUNET_OK; } -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC channel handle - */ static void -transmit_next (struct GNUNET_PSYC_Channel *ch); +handle_channel_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + struct GNUNET_PSYC_Channel *chn = cls; + uint16_t size = ntohs (res->header.size); + if (size < sizeof (*res)) + { /* Error, message too small. */ + GNUNET_break (0); + return; + } -/** - * Reset stored data related to the last received message. - */ -static void -recv_reset (struct GNUNET_PSYC_Channel *ch) -{ - 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; + uint16_t data_size = size - sizeof (*res); + const char *data = (0 < data_size) ? (void *) &res[1] : NULL; + GNUNET_OP_result (chn->op, GNUNET_ntohll (res->op_id), + GNUNET_ntohll (res->result_code), + data, data_size, NULL); } static void -recv_error (struct GNUNET_PSYC_Channel *ch) +op_recv_history_result (void *cls, int64_t result, + const void *data, uint16_t data_size) { - GNUNET_PSYC_MessageCallback message_cb - = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC - ? ch->hist_message_cb - : ch->message_cb; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Received history replay result: %" PRId64 ".\n", result); + + struct GNUNET_PSYC_HistoryRequest *hist = cls; - if (NULL != message_cb) - message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, NULL); + if (NULL != hist->result_cb) + hist->result_cb (hist->cls, result, data, data_size); - recv_reset (ch); + GNUNET_PSYC_receive_destroy (hist->recv); + GNUNET_free (hist); } -/** - * 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) +op_recv_state_result (void *cls, int64_t result, + const void *data, uint16_t data_size) { - 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); + "Received state request result: %" PRId64 ".\n", result); - 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; - } - } + struct GNUNET_PSYC_StateRequest *sr = cls; - 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 != sr->result_cb) + sr->result_cb (sr->cls, result, data, data_size); + + GNUNET_free (sr); +} - 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; +static int +check_channel_history_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + struct GNUNET_PSYC_MessageHeader * + pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res); + uint16_t size = ntohs (res->header.size); - transmit_next (ch); + if ( (NULL == pmsg) || + (size < sizeof (*res) + sizeof (*pmsg)) ) + { /* Error, message too small. */ + GNUNET_break_op (0); + return GNUNET_SYSERR; + } + return GNUNET_OK; } -/** - * Request a modifier from a client to transmit. - * - * @param mst Master handle. - */ static void -channel_transmit_mod (struct GNUNET_PSYC_Channel *ch) +handle_channel_history_result (void *cls, + const struct GNUNET_OperationResultMessage *res) { - 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; + struct GNUNET_PSYC_Channel *chn = cls; + struct GNUNET_PSYC_MessageHeader * + pmsg = (struct GNUNET_PSYC_MessageHeader *) GNUNET_MQ_extract_nested_mh (res); + GNUNET_ResultCallback result_cb = NULL; + struct GNUNET_PSYC_HistoryRequest *hist = NULL; - 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); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "%p Received historic fragment for message #%" PRIu64 ".\n", + chn, + GNUNET_ntohll (pmsg->message_id)); + + if (GNUNET_YES != GNUNET_OP_get (chn->op, + GNUNET_ntohll (res->op_id), + &result_cb, (void *) &hist, NULL)) + { /* Operation not found. */ + LOG (GNUNET_ERROR_TYPE_WARNING, + "%p Replay operation not found for historic fragment of message #%" + PRIu64 ".\n", + chn, GNUNET_ntohll (pmsg->message_id)); + return; } - 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; + GNUNET_PSYC_receive_message (hist->recv, + (const struct GNUNET_PSYC_MessageHeader *) pmsg); +} - 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)); +static int +check_channel_state_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res); + uint16_t mod_size; + uint16_t size; - queue_message (ch, msg, GNUNET_YES); - return; + if (NULL == mod) + { + GNUNET_break_op (0); + return GNUNET_SYSERR; } - - if (0 < data_size) + mod_size = ntohs (mod->size); + size = ntohs (res->header.size); + if (size - sizeof (*res) != mod_size) { - GNUNET_assert (data_size <= max_data_size); - msg->size = htons (msg->size + data_size); - queue_message (ch, msg, GNUNET_NO); + GNUNET_break_op (0); + return GNUNET_SYSERR; } - - channel_transmit_mod (ch); + return GNUNET_OK; } -/** - * Request data from a client to transmit. - * - * @param mst Master handle. - */ static void -channel_transmit_data (struct GNUNET_PSYC_Channel *ch) +handle_channel_state_result (void *cls, + const struct GNUNET_OperationResultMessage *res) { - uint16_t data_size = GNUNET_PSYC_DATA_MAX_PAYLOAD; - char data[GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD] = ""; - struct GNUNET_MessageHeader *msg = (struct GNUNET_MessageHeader *) data; + struct GNUNET_PSYC_Channel *chn = cls; + + GNUNET_ResultCallback result_cb = NULL; + struct GNUNET_PSYC_StateRequest *sr = NULL; - msg->type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA); + if (GNUNET_YES != GNUNET_OP_get (chn->op, + GNUNET_ntohll (res->op_id), + &result_cb, (void *) &sr, NULL)) + { /* Operation not found. */ + return; + } - int notify_ret = ch->tmit.notify_data (ch->tmit.notify_cls, - &data_size, &msg[1]); - switch (notify_ret) + const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res); + uint16_t mod_size = ntohs (mod->size); + + switch (ntohs (mod->type)) + { + case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER: { - case GNUNET_NO: - if (0 == data_size) + const struct GNUNET_PSYC_MessageModifier * + pmod = (const struct GNUNET_PSYC_MessageModifier *) mod; + + const char *name = (const char *) &pmod[1]; + uint16_t name_size = ntohs (pmod->name_size); + if ('\0' != name[name_size - 1]) { - /* Transmission paused, nothing to send. */ - ch->tmit_paused = GNUNET_YES; + GNUNET_break (0); return; } + sr->var_cb (sr->cls, mod, name, name + name_size, + ntohs (pmod->header.size) - sizeof (*pmod), + ntohs (pmod->value_size)); 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); + case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MOD_CONT: + sr->var_cb (sr->cls, mod, NULL, (const char *) &mod[1], + mod_size - sizeof (*mod), 0); + break; } } -/** - * 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) +static int +check_channel_message (void *cls, + const struct GNUNET_PSYC_MessageHeader *pmsg) { - 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; + return GNUNET_OK; } -/** - * 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) +handle_channel_message (void *cls, + const struct GNUNET_PSYC_MessageHeader *pmsg) { - 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_Channel *chn = cls; + + GNUNET_PSYC_receive_message (chn->recv, pmsg); } -/** - * 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) +handle_channel_message_ack (void *cls, + const struct GNUNET_MessageHeader *msg) { - struct GNUNET_PSYC_Channel *ch = th->ch; - if (GNUNET_NO == ch->in_transmit) - return; + struct GNUNET_PSYC_Channel *chn = cls; + + GNUNET_PSYC_transmit_got_ack (chn->tmit); } -/** - * 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) +handle_master_start_ack (void *cls, + const struct GNUNET_PSYC_CountersResultMessage *cres) { - 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); + struct GNUNET_PSYC_Master *mst = cls; - 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) + int32_t result = ntohl (cres->result_code); + if (GNUNET_OK != result && GNUNET_NO != result) { - // 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; + LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master: %ld\n", result); + GNUNET_break (0); + /* FIXME: disconnect */ } + if (NULL != mst->start_cb) + mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id)); +} - uint16_t pos = 0, psize = 0, ptype, size_eq, size_min; - for (pos = 0; sizeof (*msg) + pos < size; pos += psize) +static int +check_master_join_request (void *cls, + const struct GNUNET_PSYC_JoinRequestMessage *req) +{ + if ( ((sizeof (*req) + sizeof (struct GNUNET_PSYC_Message)) <= ntohs (req->header.size)) && + (NULL == GNUNET_MQ_extract_nested_mh (req)) ) { - 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); + GNUNET_break_op (0); + return GNUNET_SYSERR; + } + return GNUNET_OK; +} - 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; - } +static void +handle_master_join_request (void *cls, + const struct GNUNET_PSYC_JoinRequestMessage *req) +{ + struct GNUNET_PSYC_Master *mst = cls; - 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; - } - } + if (NULL == mst->join_req_cb) + return; - GNUNET_PSYC_MessageCallback message_cb - = ch->recv_flags & GNUNET_PSYC_MESSAGE_HISTORIC - ? ch->hist_message_cb - : ch->message_cb; + const struct GNUNET_PSYC_Message *join_msg = NULL; + if (sizeof (*req) + sizeof (*join_msg) <= ntohs (req->header.size)) + { + join_msg = (struct GNUNET_PSYC_Message *) GNUNET_MQ_extract_nested_mh (req); + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Received join_msg of type %u and size %u.\n", + ntohs (join_msg->header.type), + ntohs (join_msg->header.size)); + } - if (NULL != message_cb) - message_cb (ch->cb_cls, ch->recv_message_id, ch->recv_flags, pmsg); + struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); + jh->mst = mst; + jh->slave_pub_key = req->slave_pub_key; - 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, &req->slave_pub_key, join_msg, 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) +handle_slave_join_ack (void *cls, + const struct GNUNET_PSYC_CountersResultMessage *cres) { - if (0 == ch->tmit_ack_pending) + struct GNUNET_PSYC_Slave *slv = cls; + + int32_t result = ntohl (cres->result_code); + if (GNUNET_YES != result && GNUNET_NO != result) { - LOG (GNUNET_ERROR_TYPE_WARNING, "Ignoring extraneous message ACK\n"); + LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n"); GNUNET_break (0); - return; + /* FIXME: disconnect */ } - 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; + if (NULL != slv->connect_cb) + slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id)); +} - 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); - } +static int +check_slave_join_decision (void *cls, + const struct GNUNET_PSYC_JoinDecisionMessage *dcsn) +{ + return GNUNET_OK; } -/** - * 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) +handle_slave_join_decision (void *cls, + const struct GNUNET_PSYC_JoinDecisionMessage *dcsn) { - // YUCK! => please have disjoint message handlers... - 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); + struct GNUNET_PSYC_Message *pmsg = NULL; + if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg)) + pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1]; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Received message of type %d and size %u from PSYC service\n", - type, size); + if (NULL != slv->join_dcsn_cb) + slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg); +} - 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; - default: - GNUNET_break_op (0); - return; - } - if (! ((0 < size_eq && size == size_eq) - || (0 < size_min && size_min <= size))) +static void +channel_cleanup (struct GNUNET_PSYC_Channel *chn) +{ + if (NULL != chn->tmit) { - GNUNET_break_op (0); - return; + GNUNET_PSYC_transmit_destroy (chn->tmit); + chn->tmit = NULL; } - - switch (type) + if (NULL != chn->recv) { - 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; + GNUNET_PSYC_receive_destroy (chn->recv); + chn->recv = NULL; } - case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK: + if (NULL != chn->connect_env) { - struct CountersResult *cres = (struct CountersResult *) msg; - if (NULL != slv->join_cb) - slv->join_cb (ch->cb_cls, GNUNET_ntohll (cres->max_message_id)); - break; + GNUNET_MQ_discard (chn->connect_env); + chn->connect_env = NULL; } - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK: + if (NULL != chn->mq) { - handle_psyc_message_ack (ch); - break; - } - - case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE: - handle_psyc_message (ch, (const struct GNUNET_PSYC_MessageHeader *) msg); - break; + GNUNET_MQ_destroy (chn->mq); + chn->mq = NULL; } - - if (NULL != ch->client) + if (NULL != chn->disconnect_cb) { - GNUNET_CLIENT_receive (ch->client, &message_handler, ch, - GNUNET_TIME_UNIT_FOREVER_REL); + chn->disconnect_cb (chn->disconnect_cls); + chn->disconnect_cb = NULL; } + GNUNET_free (chn); } -/** - * 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) +static void +channel_disconnect (struct GNUNET_PSYC_Channel *chn, + GNUNET_ContinuationCallback cb, + void *cls) { - 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) + chn->is_disconnecting = GNUNET_YES; + chn->disconnect_cb = cb; + chn->disconnect_cls = cls; + + if (NULL != chn->mq) { - reschedule_connect (ch); - return 0; + struct GNUNET_MQ_Envelope *env = GNUNET_MQ_get_last_envelope (chn->mq); + if (NULL != env) + { + GNUNET_MQ_notify_sent (env, (GNUNET_MQ_NotifyCallback) channel_cleanup, chn); + } + else + { + channel_cleanup (chn); + } } - 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) + else { - ch->in_receive = GNUNET_YES; - GNUNET_CLIENT_receive (ch->client, &message_handler, ch, - GNUNET_TIME_UNIT_FOREVER_REL); + channel_cleanup (chn); } - return ret; } -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC handle. - */ +/*** MASTER ***/ + + static void -transmit_next (struct GNUNET_PSYC_Channel *ch) -{ - LOG (GNUNET_ERROR_TYPE_DEBUG, "transmit_next()\n"); - if (NULL != ch->th || NULL == ch->client) - return; +master_connect (struct GNUNET_PSYC_Master *mst); - 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); + +static void +master_reconnect (void *cls) +{ + master_connect (cls); } /** - * Try again to connect to the PSYC service. + * Master client disconnected from service. * - * @param cls Channel handle. - * @param tc Scheduler context. + * Reconnect after backoff period. */ static void -reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc) +master_disconnected (void *cls, enum GNUNET_MQ_Error error) { - struct GNUNET_PSYC_Channel *ch = cls; + struct GNUNET_PSYC_Master *mst = cls; + struct GNUNET_PSYC_Channel *chn = &mst->chn; - 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)) + "Master client disconnected (%d), re-connecting\n", + (int) error); + if (NULL != chn->tmit) { - 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); + GNUNET_PSYC_transmit_destroy (chn->tmit); + chn->tmit = NULL; } - transmit_next (ch); + if (NULL != chn->mq) + { + GNUNET_MQ_destroy (chn->mq); + chn->mq = NULL; + } + chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay, + master_reconnect, + mst); + chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay); } -/** - * Disconnect from the PSYC service. - * - * @param c Channel handle to disconnect. - */ static void -disconnect (void *c) +master_connect (struct GNUNET_PSYC_Master *mst) { - 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; - } + struct GNUNET_PSYC_Channel *chn = &mst->chn; + + struct GNUNET_MQ_MessageHandler handlers[] = { + GNUNET_MQ_hd_fixed_size (master_start_ack, + GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK, + struct GNUNET_PSYC_CountersResultMessage, + mst), + GNUNET_MQ_hd_var_size (master_join_request, + GNUNET_MESSAGE_TYPE_PSYC_JOIN_REQUEST, + struct GNUNET_PSYC_JoinRequestMessage, + mst), + GNUNET_MQ_hd_var_size (channel_message, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, + struct GNUNET_PSYC_MessageHeader, + chn), + GNUNET_MQ_hd_fixed_size (channel_message_ack, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK, + struct GNUNET_MessageHeader, + chn), + GNUNET_MQ_hd_var_size (channel_history_result, + GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_hd_var_size (channel_state_result, + GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_hd_var_size (channel_result, + GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_handler_end () + }; + + chn->mq = GNUNET_CLIENT_connecT (chn->cfg, "psyc", + handlers, master_disconnected, mst); + GNUNET_assert (NULL != chn->mq); + chn->tmit = GNUNET_PSYC_transmit_create (chn->mq); + + GNUNET_MQ_send_copy (chn->mq, chn->connect_env); } @@ -1150,20 +708,20 @@ disconnect (void *c) * or part messages, the respective methods must call other PSYC functions to * inform PSYC about the meaning of the respective events. * - * @param cfg Configuration to use (to connect to PSYC service). - * @param channel_key ECC key that will be used to sign messages for this + * @param cfg Configuration to use (to connect to PSYC service). + * @param channel_key ECC key that will be used to sign messages for this * PSYC session. The public key is used to identify the PSYC channel. * Note that end-users will usually not use the private key directly, but * rather look it up in GNS for places managed by other users, or select * a file with the private key(s) when setting up their own channels * FIXME: we'll likely want to use NOT the p521 curve here, but a cheaper * one in the future. - * @param policy Channel policy specifying join and history restrictions. + * @param policy Channel policy specifying join and history restrictions. * Used to automate join decisions. - * @param message_cb Function to invoke on message parts received from slaves. - * @param join_cb Function to invoke when a peer wants to join. - * @param master_started_cb Function to invoke after the channel master started. - * @param cls Closure for @a master_started_cb and @a join_cb. + * @param message_cb Function to invoke on message parts received from slaves. + * @param join_request_cb Function to invoke when a slave wants to join. + * @param master_start_cb Function to invoke after the channel master started. + * @param cls Closure for @a method and @a join_cb. * * @return Handle for the channel master, NULL on error. */ @@ -1171,30 +729,33 @@ struct GNUNET_PSYC_Master * GNUNET_PSYC_master_start (const struct GNUNET_CONFIGURATION_Handle *cfg, const struct GNUNET_CRYPTO_EddsaPrivateKey *channel_key, enum GNUNET_PSYC_Policy policy, + GNUNET_PSYC_MasterStartCallback start_cb, + GNUNET_PSYC_JoinRequestCallback join_request_cb, GNUNET_PSYC_MessageCallback message_cb, - GNUNET_PSYC_JoinCallback join_cb, - GNUNET_PSYC_MasterStartCallback master_started_cb, + GNUNET_PSYC_MessagePartCallback message_part_cb, 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_Master *mst = GNUNET_new (struct GNUNET_PSYC_Master); + struct GNUNET_PSYC_Channel *chn = &mst->chn; + struct MasterStartRequest *req; - req->header.size = htons (sizeof (*req)); - req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MASTER_START); + chn->connect_env = GNUNET_MQ_msg (req, + GNUNET_MESSAGE_TYPE_PSYC_MASTER_START); req->channel_key = *channel_key; req->policy = policy; - mst->start_cb = master_started_cb; - ch->message_cb = message_cb; - ch->join_cb = join_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); + chn->cfg = cfg; + chn->is_master = GNUNET_YES; + chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS; + + chn->op = GNUNET_OP_create (); + chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls); + + mst->start_cb = start_cb; + mst->join_req_cb = join_request_cb; + mst->cb_cls = cls; + master_connect (mst); return mst; } @@ -1203,12 +764,19 @@ 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, + int keep_active, + GNUNET_ContinuationCallback stop_cb, + void *stop_cls) { - disconnect (master); - GNUNET_free (master); + struct GNUNET_PSYC_Channel *chn = &mst->chn; + + /* FIXME: send msg to service */ + + channel_disconnect (chn, stop_cb, stop_cls); } @@ -1219,8 +787,9 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master) * #GNUNET_PSYC_JoinCallback. * * @param jh Join request handle. - * @param is_admitted #GNUNET_YES if joining is approved, - * #GNUNET_NO if it is disapproved. + * @param is_admitted #GNUNET_YES if the join is approved, + * #GNUNET_NO if it is disapproved, + * #GNUNET_SYSERR if we cannot answer the request. * @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 @@ -1229,22 +798,40 @@ GNUNET_PSYC_master_stop (struct GNUNET_PSYC_Master *master) * 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 join_resp Application-dependent join response message. + * + * @return #GNUNET_OK on success, + * #GNUNET_SYSERR if the message is too large. */ -void +int GNUNET_PSYC_join_decision (struct GNUNET_PSYC_JoinHandle *jh, int is_admitted, uint32_t relay_count, const struct GNUNET_PeerIdentity *relays, - const char *method_name, - const struct GNUNET_ENV_Environment *env, - const void *data, - size_t data_size) + const struct GNUNET_PSYC_Message *join_resp) { - + struct GNUNET_PSYC_Channel *chn = &jh->mst->chn; + struct GNUNET_PSYC_JoinDecisionMessage *dcsn; + uint16_t join_resp_size + = (NULL != join_resp) ? ntohs (join_resp->header.size) : 0; + uint16_t relay_size = relay_count * sizeof (*relays); + + if (GNUNET_MULTICAST_FRAGMENT_MAX_PAYLOAD + < sizeof (*dcsn) + relay_size + join_resp_size) + return GNUNET_SYSERR; + + struct GNUNET_MQ_Envelope * + env = GNUNET_MQ_msg_extra (dcsn, relay_size + join_resp_size, + GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION); + dcsn->is_admitted = htonl (is_admitted); + dcsn->slave_pub_key = jh->slave_pub_key; + + if (0 < join_resp_size) + GNUNET_memcpy (&dcsn[1], join_resp, join_resp_size); + + GNUNET_MQ_send (chn->mq, env); + GNUNET_free (jh); + return GNUNET_OK; } @@ -1261,40 +848,150 @@ 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) +{ + 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; +} + + +/*** SLAVE ***/ + + +static void +slave_connect (struct GNUNET_PSYC_Slave *slv); + + +static void +slave_reconnect (void *cls) +{ + slave_connect (cls); +} + + +/** + * Slave client disconnected from service. + * + * Reconnect after backoff period. + */ +static void +slave_disconnected (void *cls, enum GNUNET_MQ_Error error) +{ + struct GNUNET_PSYC_Slave *slv = cls; + struct GNUNET_PSYC_Channel *chn = &slv->chn; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Slave client disconnected (%d), re-connecting\n", + (int) error); + if (NULL != chn->tmit) + { + GNUNET_PSYC_transmit_destroy (chn->tmit); + chn->tmit = NULL; + } + if (NULL != chn->mq) + { + GNUNET_MQ_destroy (chn->mq); + chn->mq = NULL; + } + chn->reconnect_task = GNUNET_SCHEDULER_add_delayed (chn->reconnect_delay, + slave_reconnect, + slv); + chn->reconnect_delay = GNUNET_TIME_STD_BACKOFF (chn->reconnect_delay); +} + + +static void +slave_connect (struct GNUNET_PSYC_Slave *slv) { - channel_transmit_cancel ((struct GNUNET_PSYC_ChannelTransmitHandle *) th); + struct GNUNET_PSYC_Channel *chn = &slv->chn; + + struct GNUNET_MQ_MessageHandler handlers[] = { + GNUNET_MQ_hd_fixed_size (slave_join_ack, + GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK, + struct GNUNET_PSYC_CountersResultMessage, + slv), + GNUNET_MQ_hd_var_size (slave_join_decision, + GNUNET_MESSAGE_TYPE_PSYC_JOIN_DECISION, + struct GNUNET_PSYC_JoinDecisionMessage, + slv), + GNUNET_MQ_hd_var_size (channel_message, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE, + struct GNUNET_PSYC_MessageHeader, + chn), + GNUNET_MQ_hd_fixed_size (channel_message_ack, + GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_ACK, + struct GNUNET_MessageHeader, + chn), + GNUNET_MQ_hd_var_size (channel_history_result, + GNUNET_MESSAGE_TYPE_PSYC_HISTORY_RESULT, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_hd_var_size (channel_state_result, + GNUNET_MESSAGE_TYPE_PSYC_STATE_RESULT, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_hd_var_size (channel_result, + GNUNET_MESSAGE_TYPE_PSYC_RESULT_CODE, + struct GNUNET_OperationResultMessage, + chn), + GNUNET_MQ_handler_end () + }; + + chn->mq = GNUNET_CLIENT_connecT (chn->cfg, "psyc", + handlers, slave_disconnected, slv); + GNUNET_assert (NULL != chn->mq); + chn->tmit = GNUNET_PSYC_transmit_create (chn->mq); + + GNUNET_MQ_send_copy (chn->mq, chn->connect_env); } @@ -1308,69 +1005,82 @@ GNUNET_PSYC_master_transmit_cancel (struct GNUNET_PSYC_MasterTransmitHandle *th) * notification on failure (as the channel may simply take days to approve, * and disapproval is simply being ignored). * - * @param cfg Configuration to use. - * @param channel_key ECC public key that identifies the channel we wish to join. - * @param slave_key ECC private-public key pair that identifies the slave, and + * @param cfg + * Configuration to use. + * @param channel_key ECC public key that identifies the channel we wish to join. + * @param slave_key ECC private-public key pair that identifies the slave, and * used by multicast to sign the join request and subsequent unicast * requests sent to the master. - * @param origin Peer identity of the origin. - * @param relay_count Number of peers in the @a relays array. - * @param relays Peer identities of members of the multicast group, which serve + * @param origin Peer identity of the origin. + * @param relay_count Number of peers in the @a relays array. + * @param relays Peer identities of members of the multicast group, which serve * as relays and used to join the group at. - * @param message_cb Function to invoke on message parts received from the + * @param message_cb Function to invoke on message parts received from the * channel, typically at least contains method handlers for @e join and * @e part. - * @param join_cb function invoked once we have joined with the current - * message ID of the channel - * @param slave_joined_cb Function to invoke when a peer wants to join. - * @param cls Closure for @a message_cb and @a slave_joined_cb. - * @param method_name Method name for the join request. - * @param env Environment containing transient variables for the request, or NULL. - * @param data Payload for the join message. - * @param data_size Number of bytes in @a data. + * @param slave_connect_cb Function invoked once we have connected to the + * PSYC service. + * @param join_decision_cb Function invoked once we have received a join + * decision. + * @param cls Closure for @a message_cb and @a slave_joined_cb. + * @param method_name Method name for the join request. + * @param env Environment containing transient variables for the request, or NULL. + * @param data Payload for the join message. + * @param data_size Number of bytes in @a data. * * @return Handle for the slave, NULL on error. */ struct GNUNET_PSYC_Slave * GNUNET_PSYC_slave_join (const struct GNUNET_CONFIGURATION_Handle *cfg, - const struct GNUNET_CRYPTO_EddsaPublicKey *channel_key, - const struct GNUNET_CRYPTO_EddsaPrivateKey *slave_key, + const struct GNUNET_CRYPTO_EddsaPublicKey *channel_pub_key, + const struct GNUNET_CRYPTO_EcdsaPrivateKey *slave_key, + enum GNUNET_PSYC_SlaveJoinFlags flags, const struct GNUNET_PeerIdentity *origin, uint32_t relay_count, const struct GNUNET_PeerIdentity *relays, GNUNET_PSYC_MessageCallback message_cb, - GNUNET_PSYC_JoinCallback join_cb, - GNUNET_PSYC_SlaveJoinCallback slave_joined_cb, + GNUNET_PSYC_MessagePartCallback message_part_cb, + GNUNET_PSYC_SlaveConnectCallback connect_cb, + GNUNET_PSYC_JoinDecisionCallback join_decision_cb, void *cls, - const char *method_name, - const struct GNUNET_ENV_Environment *env, - const void *data, - uint16_t data_size) + const struct GNUNET_PSYC_Message *join_msg) { struct GNUNET_PSYC_Slave *slv = GNUNET_malloc (sizeof (*slv)); - struct GNUNET_PSYC_Channel *ch = &slv->ch; - struct SlaveJoinRequest *req - = GNUNET_malloc (sizeof (*req) + relay_count * sizeof (*relays)); - req->header.size = htons (sizeof (*req) - + relay_count * sizeof (*relays)); - req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN); - req->channel_key = *channel_key; + struct GNUNET_PSYC_Channel *chn = &slv->chn; + uint16_t relay_size = relay_count * sizeof (*relays); + uint16_t join_msg_size; + if (NULL == join_msg) + join_msg_size = 0; + else + join_msg_size = ntohs (join_msg->header.size); + + struct SlaveJoinRequest *req; + chn->connect_env = GNUNET_MQ_msg_extra (req, relay_size + join_msg_size, + GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN); + req->channel_pub_key = *channel_pub_key; req->slave_key = *slave_key; req->origin = *origin; req->relay_count = htonl (relay_count); - memcpy (&req[1], relays, relay_count * sizeof (*relays)); + req->flags = htonl (flags); + + if (0 < relay_size) + GNUNET_memcpy (&req[1], relays, relay_size); + + if (NULL != join_msg) + GNUNET_memcpy ((char *) &req[1] + relay_size, join_msg, join_msg_size); + + chn->cfg = cfg; + chn->is_master = GNUNET_NO; + chn->reconnect_delay = GNUNET_TIME_UNIT_MILLISECONDS; - slv->join_cb = slave_joined_cb; - ch->message_cb = message_cb; - ch->join_cb = join_cb; - ch->cb_cls = cls; + chn->op = GNUNET_OP_create (); + chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, 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); + slv->connect_cb = connect_cb; + slv->join_dcsn_cb = join_decision_cb; + slv->cb_cls = cls; + slave_connect (slv); return slv; } @@ -1384,10 +1094,16 @@ 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, + int keep_active, + GNUNET_ContinuationCallback part_cb, + void *part_cls) { - disconnect (slave); - GNUNET_free (slave); + struct GNUNET_PSYC_Channel *chn = &slv->chn; + + /* FIXME: send msg to service */ + + channel_disconnect (chn, part_cb, part_cls); } @@ -1405,69 +1121,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; } @@ -1487,29 +1193,40 @@ 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 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. + * @param chn + * Channel handle. + * @param slave_pub_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. + * @param result_cb + * Function to call with the result of the operation. + * The @e result_code argument is #GNUNET_OK on success, or + * #GNUNET_SYSERR on error. In case of an error, the @e data argument + * can contain an optional error message. + * @param cls + * Closure for @a result_cb. */ void -GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *channel, - const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key, +GNUNET_PSYC_channel_slave_add (struct GNUNET_PSYC_Channel *chn, + const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key, uint64_t announced_at, - uint64_t effective_since) + uint64_t effective_since, + GNUNET_ResultCallback result_cb, + void *cls) { - 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 ChannelMembershipStoreRequest *req; + struct GNUNET_MQ_Envelope * + env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE); + req->slave_pub_key = *slave_pub_key; + req->announced_at = GNUNET_htonll (announced_at); + req->effective_since = GNUNET_htonll (effective_since); + req->did_join = GNUNET_YES; + req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL)); + + GNUNET_MQ_send (chn->mq, env); } @@ -1530,75 +1247,215 @@ 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 slave_key Identity of channel slave to remove. - * @param announced_at ID of the message that announced the membership change. + * @param chn + * Channel handle. + * @param slave_pub_key + * Identity of channel slave to remove. + * @param announced_at + * ID of the message that announced the membership change. + * @param result_cb + * Function to call with the result of the operation. + * The @e result_code argument is #GNUNET_OK on success, or + * #GNUNET_SYSERR on error. In case of an error, the @e data argument + * can contain an optional error message. + * @param cls + * Closure for @a result_cb. */ void -GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *channel, - const struct GNUNET_CRYPTO_EddsaPublicKey *slave_key, - uint64_t announced_at) +GNUNET_PSYC_channel_slave_remove (struct GNUNET_PSYC_Channel *chn, + const struct GNUNET_CRYPTO_EcdsaPublicKey *slave_pub_key, + uint64_t announced_at, + GNUNET_ResultCallback result_cb, + void *cls) +{ + struct ChannelMembershipStoreRequest *req; + struct GNUNET_MQ_Envelope * + env = GNUNET_MQ_msg (req, GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_MEMBERSHIP_STORE); + req->slave_pub_key = *slave_pub_key; + req->announced_at = GNUNET_htonll (announced_at); + req->did_join = GNUNET_NO; + req->op_id = GNUNET_htonll (GNUNET_OP_add (chn->op, result_cb, cls, NULL)); + + GNUNET_MQ_send (chn->mq, env); +} + + +static struct GNUNET_PSYC_HistoryRequest * +channel_history_replay (struct GNUNET_PSYC_Channel *chn, + uint64_t start_message_id, + uint64_t end_message_id, + uint64_t message_limit, + const char *method_prefix, + uint32_t flags, + GNUNET_PSYC_MessageCallback message_cb, + GNUNET_PSYC_MessagePartCallback message_part_cb, + GNUNET_ResultCallback result_cb, + void *cls) { - 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 GNUNET_PSYC_HistoryRequestMessage *req; + struct GNUNET_PSYC_HistoryRequest *hist = GNUNET_malloc (sizeof (*hist)); + hist->chn = chn; + hist->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls); + hist->result_cb = result_cb; + hist->cls = cls; + hist->op_id = GNUNET_OP_add (chn->op, op_recv_history_result, hist, NULL); + + GNUNET_assert (NULL != method_prefix); + uint16_t method_size = strnlen (method_prefix, + GNUNET_SERVER_MAX_MESSAGE_SIZE + - sizeof (*req)) + 1; + GNUNET_assert ('\0' == method_prefix[method_size - 1]); + + struct GNUNET_MQ_Envelope * + env = GNUNET_MQ_msg_extra (req, method_size, + GNUNET_MESSAGE_TYPE_PSYC_HISTORY_REPLAY); + req->start_message_id = GNUNET_htonll (start_message_id); + req->end_message_id = GNUNET_htonll (end_message_id); + req->message_limit = GNUNET_htonll (message_limit); + req->flags = htonl (flags); + req->op_id = GNUNET_htonll (hist->op_id); + GNUNET_memcpy (&req[1], method_prefix, method_size); + + GNUNET_MQ_send (chn->mq, env); + return hist; } /** - * Request to be told the message history of the channel. + * Request to replay a part of the message history of the channel. + * + * Historic messages (but NOT the state at the time) will be replayed and given + * to the normal method handlers with a #GNUNET_PSYC_MESSAGE_HISTORIC flag set. + * + * Messages are retrieved from the local PSYCstore if available, + * otherwise requested from the network. + * + * @param channel + * Which channel should be replayed? + * @param start_message_id + * Earliest interesting point in history. + * @param end_message_id + * Last (inclusive) interesting point in history. + * @param method_prefix + * Retrieve only messages with a matching method prefix. + * @param flags + * OR'ed enum GNUNET_PSYC_HistoryReplayFlags + * @param result_cb + * Function to call when the requested history has been fully replayed. + * @param cls + * Closure for the callbacks. + * + * @return Handle to cancel history replay operation. + */ +struct GNUNET_PSYC_HistoryRequest * +GNUNET_PSYC_channel_history_replay (struct GNUNET_PSYC_Channel *chn, + uint64_t start_message_id, + uint64_t end_message_id, + const char *method_prefix, + uint32_t flags, + GNUNET_PSYC_MessageCallback message_cb, + GNUNET_PSYC_MessagePartCallback message_part_cb, + GNUNET_ResultCallback result_cb, + void *cls) +{ + return channel_history_replay (chn, start_message_id, end_message_id, 0, + method_prefix, flags, + message_cb, message_part_cb, result_cb, cls); +} + + +/** + * Request to replay the latest messages from the message history of the channel. * * Historic messages (but NOT the state at the time) will be replayed (given to * the normal method handlers) if available and if access is permitted. * - * To get the latest message, use 0 for both the start and end message ID. - * - * @param channel 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 message_cb Function to invoke on message parts received from the story. - * @param finish_cb Function to call when the requested story has been fully - * told (counting message IDs might not suffice, as some messages - * might be secret and thus the listener would not know the story is - * finished without being told explicitly) once this function - * has been called, the client must not call - * GNUNET_PSYC_channel_story_tell_cancel() anymore. - * @param cls Closure for the callbacks. - * - * @return Handle to cancel story telling operation. + * @param channel + * Which channel should be replayed? + * @param message_limit + * Maximum number of messages to replay. + * @param method_prefix + * Retrieve only messages with a matching method prefix. + * Use NULL or "" to retrieve all. + * @param flags + * OR'ed enum GNUNET_PSYC_HistoryReplayFlags + * @param result_cb + * Function to call when the requested history has been fully replayed. + * @param cls + * Closure for the callbacks. + * + * @return Handle to cancel history replay operation. */ -struct GNUNET_PSYC_Story * -GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel, - uint64_t start_message_id, - uint64_t end_message_id, - GNUNET_PSYC_MessageCallback message_cb, - GNUNET_PSYC_FinishCallback finish_cb, - void *cls) +struct GNUNET_PSYC_HistoryRequest * +GNUNET_PSYC_channel_history_replay_latest (struct GNUNET_PSYC_Channel *chn, + uint64_t message_limit, + const char *method_prefix, + uint32_t flags, + GNUNET_PSYC_MessageCallback message_cb, + GNUNET_PSYC_MessagePartCallback message_part_cb, + GNUNET_ResultCallback result_cb, + void *cls) { - return NULL; + return channel_history_replay (chn, 0, 0, message_limit, method_prefix, flags, + message_cb, message_part_cb, result_cb, cls); +} + + +void +GNUNET_PSYC_channel_history_replay_cancel (struct GNUNET_PSYC_Channel *channel, + struct GNUNET_PSYC_HistoryRequest *hist) +{ + GNUNET_PSYC_receive_destroy (hist->recv); + GNUNET_OP_remove (hist->chn->op, hist->op_id); + GNUNET_free (hist); } /** - * Abort story telling. + * Retrieve the best matching channel state variable. * - * This function must not be called from within method handlers (as given to - * GNUNET_PSYC_slave_join()) of the slave. + * If the requested variable name is not present in the state, the nearest + * less-specific name is matched; for example, requesting "_a_b" will match "_a" + * if "_a_b" does not exist. * - * @param story Story telling operation to stop. + * @param channel + * Channel handle. + * @param full_name + * Full name of the requested variable. + * The actual variable returned might have a shorter name. + * @param var_cb + * Function called once when a matching state variable is found. + * Not called if there's no matching state variable. + * @param result_cb + * Function called after the operation finished. + * (i.e. all state variables have been returned via @a state_cb) + * @param cls + * Closure for the callbacks. */ -void -GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story) +static struct GNUNET_PSYC_StateRequest * +channel_state_get (struct GNUNET_PSYC_Channel *chn, + uint16_t type, const char *name, + GNUNET_PSYC_StateVarCallback var_cb, + GNUNET_ResultCallback result_cb, void *cls) { - + struct StateRequest *req; + struct GNUNET_PSYC_StateRequest *sr = GNUNET_malloc (sizeof (*sr)); + sr->chn = chn; + sr->var_cb = var_cb; + sr->result_cb = result_cb; + sr->cls = cls; + sr->op_id = GNUNET_OP_add (chn->op, op_recv_state_result, sr, NULL); + + GNUNET_assert (NULL != name); + size_t name_size = strnlen (name, GNUNET_SERVER_MAX_MESSAGE_SIZE + - sizeof (*req)) + 1; + struct GNUNET_MQ_Envelope * + env = GNUNET_MQ_msg_extra (req, name_size, type); + req->op_id = GNUNET_htonll (sr->op_id); + GNUNET_memcpy (&req[1], name, name_size); + + GNUNET_MQ_send (chn->mq, env); + return sr; } @@ -1609,22 +1466,30 @@ GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story) * less-specific name is matched; for example, requesting "_a_b" will match "_a" * if "_a_b" does not exist. * - * @param channel Channel handle. - * @param full_name Full name of the requested variable, the actual variable - * returned might have a shorter name.. - * @param cb Function called once when a matching state variable is found. + * @param channel + * Channel handle. + * @param full_name + * Full name of the requested variable. + * The actual variable returned might have a shorter name. + * @param var_cb + * Function called once when a matching state variable is found. * Not called if there's no matching state variable. - * @param cb_cls Closure for the callbacks. - * - * @return Handle that can be used to cancel the query operation. + * @param result_cb + * Function called after the operation finished. + * (i.e. all state variables have been returned via @a state_cb) + * @param cls + * Closure for the callbacks. */ -struct GNUNET_PSYC_StateQuery * -GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel, +struct GNUNET_PSYC_StateRequest * +GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *chn, const char *full_name, - GNUNET_PSYC_StateCallback cb, - void *cb_cls) + GNUNET_PSYC_StateVarCallback var_cb, + GNUNET_ResultCallback result_cb, + void *cls) { - return NULL; + return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET, + full_name, var_cb, result_cb, cls); + } @@ -1638,33 +1503,42 @@ GNUNET_PSYC_channel_state_get (struct GNUNET_PSYC_Channel *channel, * The @a state_cb is invoked on all matching state variables asynchronously, as * the state is stored in and retrieved from the PSYCstore, * - * @param channel Channel handle. - * @param name_prefix Prefix of the state variable name to match. - * @param cb Function to call with the matching state variables. - * @param cb_cls Closure for the callbacks. - * - * @return Handle that can be used to cancel the query operation. + * @param channel + * Channel handle. + * @param name_prefix + * Prefix of the state variable name to match. + * @param var_cb + * Function called once when a matching state variable is found. + * Not called if there's no matching state variable. + * @param result_cb + * Function called after the operation finished. + * (i.e. all state variables have been returned via @a state_cb) + * @param cls + * Closure for the callbacks. */ -struct GNUNET_PSYC_StateQuery * -GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *channel, +struct GNUNET_PSYC_StateRequest * +GNUNET_PSYC_channel_state_get_prefix (struct GNUNET_PSYC_Channel *chn, const char *name_prefix, - GNUNET_PSYC_StateCallback cb, - void *cb_cls) + GNUNET_PSYC_StateVarCallback var_cb, + GNUNET_ResultCallback result_cb, + void *cls) { - return NULL; + return channel_state_get (chn, GNUNET_MESSAGE_TYPE_PSYC_STATE_GET_PREFIX, + name_prefix, var_cb, result_cb, cls); } /** - * Cancel a state query operation. + * Cancel a state request operation. * - * @param query Handle for the operation to cancel. + * @param sr + * Handle for the operation to cancel. */ void -GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateQuery *query) +GNUNET_PSYC_channel_state_get_cancel (struct GNUNET_PSYC_StateRequest *sr) { - + GNUNET_OP_remove (sr->chn->op, sr->op_id); + GNUNET_free (sr); } - /* end of psyc_api.c */