X-Git-Url: https://git.librecmc.org/?a=blobdiff_plain;f=src%2Fpsyc%2Fpsyc_api.c;h=9c58322934fa1abfc9d953b6493fdac3e1b171ad;hb=7afd744770424ea38ceb8aa41b9eb94f5c089f87;hp=51902dd1a09110908e7de18c898f11ab996c0be8;hpb=c654ca8072bc5768bd01c6b0d6db303b0d4e5025;p=oweals%2Fgnunet.git diff --git a/src/psyc/psyc_api.c b/src/psyc/psyc_api.c index 51902dd1a..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. */ /** @@ -30,22 +30,18 @@ * @author Gabor X Toth */ +#include + #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 OperationHandle -{ - struct OperationHandle *prev; - struct OperationHandle *next; - const struct GNUNET_MessageHeader *msg; -}; - /** * Handle to access PSYC channel operations for both the master and slaves. */ @@ -57,77 +53,65 @@ struct GNUNET_PSYC_Channel 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 operations to transmit. + * Time to wait until we try to reconnect on failure. */ - struct OperationHandle *transmit_head; + struct GNUNET_TIME_Relative reconnect_delay; /** - * Tail of operations to transmit. + * Task for reconnecting when the listener fails. */ - struct OperationHandle *transmit_tail; + struct GNUNET_SCHEDULER_Task *reconnect_task; /** - * Message to send on reconnect. + * Async operations. */ - struct GNUNET_MessageHeader *reconnect_msg; + struct GNUNET_OP_Handle *op; /** - * Task doing exponential back-off trying to reconnect. + * Transmission handle; */ - GNUNET_SCHEDULER_TaskIdentifier reconnect_task; + struct GNUNET_PSYC_TransmitHandle *tmit; /** - * Time for next connect retry. + * Receipt handle; */ - struct GNUNET_TIME_Relative reconnect_delay; + struct GNUNET_PSYC_ReceiveHandle *recv; - GNUNET_PSYC_Method method_cb; - - GNUNET_PSYC_JoinCallback join_cb; - - void *cb_cls; + /** + * Function called after disconnected from the service. + */ + GNUNET_ContinuationCallback disconnect_cb; /** - * Are we polling for incoming messages right now? + * Closure for @a disconnect_cb. */ - int in_receive; + void *disconnect_cls; /** - * Are we currently transmitting a message? + * Are we polling for incoming messages right now? */ - int in_transmit; + uint8_t in_receive; /** * Is this a master or slave channel? */ - int is_master; + uint8_t is_master; /** - * Buffer space available for transmitting the next data fragment. + * Is this channel in the process of disconnecting from the service? + * #GNUNET_YES or #GNUNET_NO */ - uint16_t tmit_buf_avail; -}; - - -/** - * Handle for a pending PSYC transmission operation. - */ -struct GNUNET_PSYC_MasterTransmitHandle -{ - struct GNUNET_PSYC_Master *master; - GNUNET_PSYC_MasterTransmitNotify notify; - void *notify_cls; - enum GNUNET_PSYC_DataStatus status; + uint8_t is_disconnecting; }; @@ -136,13 +120,19 @@ struct GNUNET_PSYC_MasterTransmitHandle */ struct GNUNET_PSYC_Master { - struct GNUNET_PSYC_Channel ch; - - struct GNUNET_PSYC_MasterTransmitHandle *tmit; + struct GNUNET_PSYC_Channel chn; GNUNET_PSYC_MasterStartCallback start_cb; - uint64_t max_message_id; + /** + * Join request callback. + */ + GNUNET_PSYC_JoinRequestCallback join_req_cb; + + /** + * Closure for the callbacks. + */ + void *cb_cls; }; @@ -151,19 +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_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; }; @@ -176,352 +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; + + /** + * Function to call when the operation finished. + */ + GNUNET_ResultCallback result_cb; + /** + * Closure for @a result_cb. + */ + void *cls; }; -/** - * Handle for a state query operation. - */ -struct GNUNET_PSYC_StateQuery +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; + + /** + * Function to call when the operation finished. + */ + GNUNET_ResultCallback result_cb; + + /** + * Closure for @a result_cb. + */ + void *cls; }; -/** - * Try again to connect to the PSYC service. - * - * @param cls Handle to the PSYC service. - * @param tc Scheduler context - */ -static void -reconnect (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc); +static int +check_channel_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + return GNUNET_OK; +} -/** - * Reschedule a connect attempt to the service. - * - * @param c channel to reconnect - */ static void -reschedule_connect (struct GNUNET_PSYC_Channel *c) +handle_channel_result (void *cls, + const struct GNUNET_OperationResultMessage *res) { - GNUNET_assert (c->reconnect_task == GNUNET_SCHEDULER_NO_TASK); + struct GNUNET_PSYC_Channel *chn = cls; - if (NULL != c->th) - { - GNUNET_CLIENT_notify_transmit_ready_cancel (c->th); - c->th = NULL; - } - if (NULL != c->client) - { - GNUNET_CLIENT_disconnect (c->client); - c->client = NULL; + uint16_t size = ntohs (res->header.size); + if (size < sizeof (*res)) + { /* Error, message too small. */ + GNUNET_break (0); + return; } - c->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 (c->reconnect_delay, GNUNET_YES)); - c->reconnect_task = - GNUNET_SCHEDULER_add_delayed (c->reconnect_delay, &reconnect, c); - c->reconnect_delay = GNUNET_TIME_STD_BACKOFF (c->reconnect_delay); + + 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); } -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC channel handle - */ static void -transmit_next (struct GNUNET_PSYC_Channel *ch); +op_recv_history_result (void *cls, int64_t result, + const void *data, uint16_t data_size) +{ + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Received history replay result: %" PRId64 ".\n", result); + + struct GNUNET_PSYC_HistoryRequest *hist = cls; + + if (NULL != hist->result_cb) + hist->result_cb (hist->cls, result, data, data_size); + + GNUNET_PSYC_receive_destroy (hist->recv); + GNUNET_free (hist); +} static void -master_transmit_data (struct GNUNET_PSYC_Master *mst) +op_recv_state_result (void *cls, int64_t result, + const void *data, uint16_t data_size) { - struct GNUNET_PSYC_Channel *ch = &mst->ch; - size_t data_size = ch->tmit_buf_avail; - struct GNUNET_PSYC_MessageData *pdata; - struct OperationHandle *op - = GNUNET_malloc (sizeof (*op) + sizeof (*pdata) + data_size); - pdata = (struct GNUNET_PSYC_MessageData *) &op[1]; - op->msg = (struct GNUNET_MessageHeader *) pdata; - pdata->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_DATA); - - switch (mst->tmit->notify (mst->tmit->notify_cls, &data_size, &pdata[1])) - { - case GNUNET_NO: - mst->tmit->status = GNUNET_PSYC_DATA_CONT; - break; + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Received state request result: %" PRId64 ".\n", result); + + struct GNUNET_PSYC_StateRequest *sr = cls; + + if (NULL != sr->result_cb) + sr->result_cb (sr->cls, result, data, data_size); + + GNUNET_free (sr); +} - case GNUNET_YES: - mst->tmit->status = GNUNET_PSYC_DATA_END; - break; - default: - mst->tmit->status = GNUNET_PSYC_DATA_CANCEL; - data_size = 0; - LOG (GNUNET_ERROR_TYPE_ERROR, "MasterTransmitNotify returned error\n"); +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); + + if ( (NULL == pmsg) || + (size < sizeof (*res) + sizeof (*pmsg)) ) + { /* Error, message too small. */ + GNUNET_break_op (0); + return GNUNET_SYSERR; } + return GNUNET_OK; +} + + +static void +handle_channel_history_result (void *cls, + const struct GNUNET_OperationResultMessage *res) +{ + 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; + + 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; + } + + GNUNET_PSYC_receive_message (hist->recv, + (const struct GNUNET_PSYC_MessageHeader *) pmsg); +} + + +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; - if ((GNUNET_PSYC_DATA_CONT == mst->tmit->status && 0 == data_size)) + if (NULL == mod) { - /* Transmission paused, nothing to send. */ - GNUNET_free (op); + GNUNET_break_op (0); + return GNUNET_SYSERR; } - else + mod_size = ntohs (mod->size); + size = ntohs (res->header.size); + if (size - sizeof (*res) != mod_size) { - GNUNET_assert (data_size <= ch->tmit_buf_avail); - pdata->header.size = htons (sizeof (*pdata) + data_size); - pdata->status = htons (mst->tmit->status); - GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op); - transmit_next (ch); + GNUNET_break_op (0); + return GNUNET_SYSERR; } + 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_channel_state_result (void *cls, + const struct GNUNET_OperationResultMessage *res) { - // YUCK! => please have disjoint message handlers... - struct GNUNET_PSYC_Channel *ch = cls; - struct GNUNET_PSYC_Master *mst = cls; - struct GNUNET_PSYC_Slave *slv = cls; - struct CountersResult *cres; - struct TransmitAck *tack; + struct GNUNET_PSYC_Channel *chn = cls; + GNUNET_ResultCallback result_cb = NULL; + struct GNUNET_PSYC_StateRequest *sr = NULL; - if (NULL == msg) - { - reschedule_connect (ch); + if (GNUNET_YES != GNUNET_OP_get (chn->op, + GNUNET_ntohll (res->op_id), + &result_cb, (void *) &sr, NULL)) + { /* Operation not found. */ 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 from PSYC service\n", type); + const struct GNUNET_MessageHeader *mod = GNUNET_MQ_extract_nested_mh (res); + uint16_t mod_size = ntohs (mod->size); - switch (type) + switch (ntohs (mod->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_TRANSMIT_ACK: - size_eq = sizeof (struct TransmitAck); + case GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER: + { + 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]) + { + 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; } - if (! ((0 < size_eq && size == size_eq) - || (0 < size_min && size >= size_min))) - { - GNUNET_break (0); - reschedule_connect (ch); - return; + 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; } +} - switch (type) - { - case GNUNET_MESSAGE_TYPE_PSYC_MASTER_START_ACK: - cres = (struct CountersResult *) msg; - mst->max_message_id = GNUNET_ntohll (cres->max_message_id); - if (NULL != mst->start_cb) - mst->start_cb (ch->cb_cls, mst->max_message_id); - break; - case GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN_ACK: - cres = (struct CountersResult *) msg; -#if TODO - slv->max_message_id = GNUNET_ntohll (cres->max_message_id); - if (NULL != slv->join_ack_cb) - mst->join_ack_cb (ch->cb_cls, mst->max_message_id); -#endif - break; +static int +check_channel_message (void *cls, + const struct GNUNET_PSYC_MessageHeader *pmsg) +{ + return GNUNET_OK; +} - case GNUNET_MESSAGE_TYPE_PSYC_TRANSMIT_ACK: - tack = (struct TransmitAck *) msg; - if (ch->is_master) - { - GNUNET_assert (NULL != mst->tmit); - if (GNUNET_PSYC_DATA_CONT != mst->tmit->status - || NULL == mst->tmit->notify) - { - GNUNET_free (mst->tmit); - mst->tmit = NULL; - } - else - { - ch->tmit_buf_avail = ntohs (tack->buf_avail); - master_transmit_data (mst); - } - } - else - { - /* TODO: slave */ - } - break; - } - GNUNET_CLIENT_receive (ch->client, &message_handler, ch, - GNUNET_TIME_UNIT_FOREVER_REL); +static void +handle_channel_message (void *cls, + const struct GNUNET_PSYC_MessageHeader *pmsg) +{ + struct GNUNET_PSYC_Channel *chn = cls; + + GNUNET_PSYC_receive_message (chn->recv, pmsg); } -/** - * Transmit next message to service. - * - * @param cls The 'struct GNUNET_PSYC_Channel'. - * @param size Number of bytes available in buf. - * @param buf Where to copy the message. - * @return Number of bytes copied to buf. - */ -static size_t -send_next_message (void *cls, size_t size, void *buf) +static void +handle_channel_message_ack (void *cls, + const struct GNUNET_MessageHeader *msg) { - struct GNUNET_PSYC_Channel *ch = cls; - struct OperationHandle *op = ch->transmit_head; - size_t ret; - - ch->th = NULL; - if (NULL == op->msg) - return 0; - ret = ntohs (op->msg->size); - if (ret > size) + struct GNUNET_PSYC_Channel *chn = cls; + + GNUNET_PSYC_transmit_got_ack (chn->tmit); +} + + +static void +handle_master_start_ack (void *cls, + const struct GNUNET_PSYC_CountersResultMessage *cres) +{ + struct GNUNET_PSYC_Master *mst = cls; + + int32_t result = ntohl (cres->result_code); + if (GNUNET_OK != result && GNUNET_NO != result) { - reschedule_connect (ch); - return 0; + LOG (GNUNET_ERROR_TYPE_ERROR, "Could not start master: %ld\n", result); + GNUNET_break (0); + /* FIXME: disconnect */ } - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Sending message of type %d to PSYC service\n", - ntohs (op->msg->type)); - memcpy (buf, op->msg, ret); - - GNUNET_CONTAINER_DLL_remove (ch->transmit_head, ch->transmit_tail, op); - GNUNET_free (op); + if (NULL != mst->start_cb) + mst->start_cb (mst->cb_cls, result, GNUNET_ntohll (cres->max_message_id)); +} - if (NULL != ch->transmit_head) - transmit_next (ch); - if (GNUNET_NO == ch->in_receive) +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)) ) { - ch->in_receive = GNUNET_YES; - GNUNET_CLIENT_receive (ch->client, &message_handler, ch, - GNUNET_TIME_UNIT_FOREVER_REL); + GNUNET_break_op (0); + return GNUNET_SYSERR; } - return ret; + return GNUNET_OK; } -/** - * Schedule transmission of the next message from our queue. - * - * @param ch PSYC handle. - */ static void -transmit_next (struct GNUNET_PSYC_Channel *ch) +handle_master_join_request (void *cls, + const struct GNUNET_PSYC_JoinRequestMessage *req) { - if (NULL != ch->th || NULL == ch->client) - return; + struct GNUNET_PSYC_Master *mst = cls; - struct OperationHandle *op = ch->transmit_head; - if (NULL == op) + if (NULL == mst->join_req_cb) return; - ch->th = GNUNET_CLIENT_notify_transmit_ready (ch->client, - ntohs (op->msg->size), - GNUNET_TIME_UNIT_FOREVER_REL, - GNUNET_NO, - &send_next_message, - ch); + 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)); + } + + struct GNUNET_PSYC_JoinHandle *jh = GNUNET_malloc (sizeof (*jh)); + jh->mst = mst; + jh->slave_pub_key = req->slave_pub_key; + + if (NULL != mst->join_req_cb) + mst->join_req_cb (mst->cb_cls, req, &req->slave_pub_key, join_msg, jh); } -/** - * 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) +handle_slave_join_ack (void *cls, + const struct GNUNET_PSYC_CountersResultMessage *cres) { - struct GNUNET_PSYC_Channel *ch = cls; - - ch->reconnect_task = GNUNET_SCHEDULER_NO_TASK; - LOG (GNUNET_ERROR_TYPE_DEBUG, - "Connecting to PSYC service.\n"); - GNUNET_assert (NULL == ch->client); - ch->client = GNUNET_CLIENT_connect ("psyc", ch->cfg); - GNUNET_assert (NULL != ch->client); + struct GNUNET_PSYC_Slave *slv = cls; - if (NULL == ch->transmit_head || - ch->transmit_head->msg->type != ch->reconnect_msg->type) + int32_t result = ntohl (cres->result_code); + if (GNUNET_YES != result && GNUNET_NO != result) { - uint16_t reconn_size = ntohs (ch->reconnect_msg->size); - struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + reconn_size); - memcpy (&op[1], ch->reconnect_msg, reconn_size); - op->msg = (struct GNUNET_MessageHeader *) &op[1]; - GNUNET_CONTAINER_DLL_insert (ch->transmit_head, ch->transmit_tail, op); + LOG (GNUNET_ERROR_TYPE_ERROR, "Could not join slave.\n"); + GNUNET_break (0); + /* FIXME: disconnect */ } - transmit_next (ch); + if (NULL != slv->connect_cb) + slv->connect_cb (slv->cb_cls, result, GNUNET_ntohll (cres->max_message_id)); +} + + +static int +check_slave_join_decision (void *cls, + const struct GNUNET_PSYC_JoinDecisionMessage *dcsn) +{ + return GNUNET_OK; } -/** - * Disconnect from the PSYC service. - * - * @param c Channel handle to disconnect - */ static void -disconnect (void *c) +handle_slave_join_decision (void *cls, + const struct GNUNET_PSYC_JoinDecisionMessage *dcsn) { - struct GNUNET_PSYC_Channel *ch = c; + struct GNUNET_PSYC_Slave *slv = cls; + + struct GNUNET_PSYC_Message *pmsg = NULL; + if (ntohs (dcsn->header.size) <= sizeof (*dcsn) + sizeof (*pmsg)) + pmsg = (struct GNUNET_PSYC_Message *) &dcsn[1]; + + if (NULL != slv->join_dcsn_cb) + slv->join_dcsn_cb (slv->cb_cls, dcsn, ntohl (dcsn->is_admitted), pmsg); +} + - GNUNET_assert (NULL != ch); - if (ch->transmit_head != ch->transmit_tail) +static void +channel_cleanup (struct GNUNET_PSYC_Channel *chn) +{ + if (NULL != chn->tmit) { - LOG (GNUNET_ERROR_TYPE_ERROR, - "Disconnecting while there are still outstanding messages!\n"); - GNUNET_break (0); + GNUNET_PSYC_transmit_destroy (chn->tmit); + chn->tmit = NULL; + } + if (NULL != chn->recv) + { + GNUNET_PSYC_receive_destroy (chn->recv); + chn->recv = NULL; + } + if (NULL != chn->connect_env) + { + GNUNET_MQ_discard (chn->connect_env); + chn->connect_env = NULL; + } + if (NULL != chn->mq) + { + GNUNET_MQ_destroy (chn->mq); + chn->mq = NULL; + } + if (NULL != chn->disconnect_cb) + { + chn->disconnect_cb (chn->disconnect_cls); + chn->disconnect_cb = NULL; } - if (ch->reconnect_task != GNUNET_SCHEDULER_NO_TASK) + GNUNET_free (chn); +} + + +static void +channel_disconnect (struct GNUNET_PSYC_Channel *chn, + GNUNET_ContinuationCallback cb, + void *cls) +{ + chn->is_disconnecting = GNUNET_YES; + chn->disconnect_cb = cb; + chn->disconnect_cls = cls; + + if (NULL != chn->mq) { - GNUNET_SCHEDULER_cancel (ch->reconnect_task); - ch->reconnect_task = GNUNET_SCHEDULER_NO_TASK; + 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); + } } - if (NULL != ch->th) + else { - GNUNET_CLIENT_notify_transmit_ready_cancel (ch->th); - ch->th = NULL; + channel_cleanup (chn); } - if (NULL != ch->client) +} + + +/*** MASTER ***/ + + +static void +master_connect (struct GNUNET_PSYC_Master *mst); + + +static void +master_reconnect (void *cls) +{ + master_connect (cls); +} + + +/** + * Master client disconnected from service. + * + * Reconnect after backoff period. + */ +static void +master_disconnected (void *cls, enum GNUNET_MQ_Error error) +{ + struct GNUNET_PSYC_Master *mst = cls; + struct GNUNET_PSYC_Channel *chn = &mst->chn; + + LOG (GNUNET_ERROR_TYPE_DEBUG, + "Master client disconnected (%d), re-connecting\n", + (int) error); + if (NULL != chn->tmit) { - GNUNET_CLIENT_disconnect (ch->client); - ch->client = NULL; + GNUNET_PSYC_transmit_destroy (chn->tmit); + chn->tmit = NULL; } - if (NULL != ch->reconnect_msg) + if (NULL != chn->mq) { - GNUNET_free (ch->reconnect_msg); - ch->reconnect_msg = NULL; + 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); +} + + +static void +master_connect (struct GNUNET_PSYC_Master *mst) +{ + 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); } @@ -537,51 +708,54 @@ 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 method Function to invoke on messages 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. */ 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_Method method, - GNUNET_PSYC_JoinCallback join_cb, - GNUNET_PSYC_MasterStartCallback master_started_cb, + GNUNET_PSYC_MasterStartCallback start_cb, + GNUNET_PSYC_JoinRequestCallback join_request_cb, + GNUNET_PSYC_MessageCallback message_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; - 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); - ch->method_cb = method; - ch->join_cb = join_cb; - ch->cb_cls = cls; - mst->start_cb = master_started_cb; + mst->start_cb = start_cb; + mst->join_req_cb = join_request_cb; + mst->cb_cls = cls; + master_connect (mst); return mst; } @@ -590,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); } @@ -606,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 @@ -616,46 +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) -{ - -} - - -/* FIXME: split up value into <64K chunks and transmit the continuations in - * MOD_CONT msgs */ -static int -send_modifier (void *cls, struct GNUNET_ENV_Modifier *mod) + const struct GNUNET_PSYC_Message *join_resp) { - struct GNUNET_PSYC_Channel *ch = cls; - size_t name_size = strlen (mod->name) + 1; - struct GNUNET_PSYC_MessageModifier *pmod; - struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*pmod) - + name_size + mod->value_size); - pmod = (struct GNUNET_PSYC_MessageModifier *) &op[1]; - op->msg = (struct GNUNET_MessageHeader *) pmod; - - pmod->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_MODIFIER); - pmod->header.size = htons (sizeof (*pmod) + name_size + mod->value_size); - pmod->name_size = htons (name_size); - memcpy (&pmod[1], mod->name, name_size); - memcpy ((char *) &pmod[1] + name_size, mod->value, mod->value_size); - - GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op); - return GNUNET_YES; + 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; } @@ -664,78 +840,158 @@ send_modifier (void *cls, struct GNUNET_ENV_Modifier *mod) * * @param master Handle to the PSYC channel. * @param method_name Which method should be invoked. - * @param env Environment containing state operations and transient variables - * for the message, or NULL. - * @param notify Function to call to obtain the arguments. - * @param notify_cls Closure for @a notify. + * @param 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). + * + * @return Transmission handle, NULL on error (i.e. more than one request queued). */ struct GNUNET_PSYC_MasterTransmitHandle * -GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *master, +GNUNET_PSYC_master_transmit (struct GNUNET_PSYC_Master *mst, const char *method_name, - const struct GNUNET_ENV_Environment *env, - GNUNET_PSYC_MasterTransmitNotify notify, + GNUNET_PSYC_TransmitNotifyModifier notify_mod, + GNUNET_PSYC_TransmitNotifyData notify_data, void *notify_cls, enum GNUNET_PSYC_MasterTransmitFlags flags) { - GNUNET_assert (NULL != master); - struct GNUNET_PSYC_Channel *ch = &master->ch; - if (GNUNET_NO != ch->in_transmit) + 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; - ch->in_transmit = GNUNET_YES; - - size_t size = strlen (method_name) + 1; - struct GNUNET_PSYC_MessageMethod *pmeth; - struct OperationHandle *op - = GNUNET_malloc (sizeof (*op) + sizeof (*pmeth) + size); - pmeth = (struct GNUNET_PSYC_MessageMethod *) &op[1]; - op->msg = (struct GNUNET_MessageHeader *) pmeth; - - pmeth->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_MESSAGE_METHOD); - pmeth->header.size = htons (sizeof (*pmeth) + size); - pmeth->flags = htonl (flags); - pmeth->mod_count = GNUNET_ntohll (GNUNET_ENV_environment_get_mod_count (env)); - memcpy (&pmeth[1], method_name, size); - - GNUNET_CONTAINER_DLL_insert_tail (ch->transmit_head, ch->transmit_tail, op); - GNUNET_ENV_environment_iterate (env, send_modifier, master); - transmit_next (ch); - - master->tmit = GNUNET_malloc (sizeof (*master->tmit)); - master->tmit->master = master; - master->tmit->notify = notify; - master->tmit->notify_cls = notify_cls; - master->tmit->status = GNUNET_PSYC_DATA_CONT; - return master->tmit; } /** * 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) { - master_transmit_data (th->master); + 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) { - struct GNUNET_PSYC_Master *master = th->master; - struct GNUNET_PSYC_Channel *ch = &master->ch; - if (GNUNET_NO != ch->in_transmit) - return; + 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) +{ + 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); } @@ -749,63 +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 method Function to invoke on messages received from the channel, - * typically at least contains functions 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 method_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 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 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_Method method, - GNUNET_PSYC_JoinCallback join_cb, - GNUNET_PSYC_SlaveJoinCallback slave_joined_cb, + GNUNET_PSYC_MessageCallback message_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)); + 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); - req->header.size = htons (sizeof (*req) - + sizeof (*channel_key) + sizeof (*slave_key) - + relay_count * sizeof (*relays)); - req->header.type = htons (GNUNET_MESSAGE_TYPE_PSYC_SLAVE_JOIN); - req->channel_key = *channel_key; + 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 = relay_count; - memcpy (&req[1], relays, relay_count * sizeof (*relays)); + req->relay_count = htonl (relay_count); + req->flags = htonl (flags); + + if (0 < relay_size) + GNUNET_memcpy (&req[1], relays, relay_size); - 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); + 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; + + chn->op = GNUNET_OP_create (); + chn->recv = GNUNET_PSYC_receive_create (message_cb, message_part_cb, cls); + + slv->connect_cb = connect_cb; + slv->join_dcsn_cb = join_decision_cb; + slv->cb_cls = cls; + + slave_connect (slv); return slv; } @@ -819,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); } @@ -831,74 +1112,68 @@ GNUNET_PSYC_slave_part (struct GNUNET_PSYC_Slave *slave) * * @param slave Slave handle. * @param method_name Which (PSYC) method should be invoked (on host). - * @param env Environment containing transient variables for the message, or - * NULL. - * @param notify Function to call when we are allowed to transmit (to get data). + * @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. * @param flags Flags for the message being transmitted. + * * @return Transmission handle, NULL on error (i.e. more than one request * queued). */ struct GNUNET_PSYC_SlaveTransmitHandle * -GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slave, +GNUNET_PSYC_slave_transmit (struct GNUNET_PSYC_Slave *slv, const char *method_name, - const struct GNUNET_ENV_Environment *env, - GNUNET_PSYC_SlaveTransmitNotify notify, + GNUNET_PSYC_TransmitNotifyModifier notify_mod, + GNUNET_PSYC_TransmitNotifyData notify_data, void *notify_cls, enum GNUNET_PSYC_SlaveTransmitFlags flags) + { - return NULL; + 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_MasterTransmitHandle *th) +GNUNET_PSYC_slave_transmit_resume (struct GNUNET_PSYC_SlaveTransmitHandle *tmit) { - + 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) { - -} - - -/** - * 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 (struct GNUNET_PSYC_Channel *) master; + 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 (struct GNUNET_PSYC_Channel *) slave; + return &slv->chn; } @@ -918,31 +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 OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvadd)); - - slvadd = (struct ChannelSlaveAdd *) &op[1]; - op->msg = (struct GNUNET_MessageHeader *) slvadd; - - slvadd->header.type = GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_ADD; - slvadd->header.size = htons (sizeof (*slvadd)); - slvadd->announced_at = GNUNET_htonll (announced_at); - slvadd->effective_since = GNUNET_htonll (effective_since); - GNUNET_CONTAINER_DLL_insert_tail (channel->transmit_head, - channel->transmit_tail, - op); - 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); } @@ -963,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 ChannelSlaveRemove *slvrm; - struct OperationHandle *op = GNUNET_malloc (sizeof (*op) + sizeof (*slvrm)); - - slvrm = (struct ChannelSlaveRemove *) &op[1]; - op->msg = (struct GNUNET_MessageHeader *) slvrm; - slvrm->header.type = GNUNET_MESSAGE_TYPE_PSYC_CHANNEL_SLAVE_RM; - slvrm->header.size = htons (sizeof (*slvrm)); - slvrm->announced_at = GNUNET_htonll (announced_at); - GNUNET_CONTAINER_DLL_insert_tail (channel->transmit_head, - channel->transmit_tail, - op); - 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->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 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 (given to - * the normal method handlers) if available and if access is permitted. + * 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. * - * To get the latest message, use 0 for both the start and end message ID. + * 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 (exclusive) interesting point in history. - * @param method Function to invoke on messages 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 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_Story * -GNUNET_PSYC_channel_story_tell (struct GNUNET_PSYC_Channel *channel, - uint64_t start_message_id, - uint64_t end_message_id, - GNUNET_PSYC_Method method, - GNUNET_PSYC_FinishCallback *finish_cb, - void *cls) +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 NULL; + return channel_history_replay (chn, start_message_id, end_message_id, 0, + method_prefix, flags, + message_cb, message_part_cb, result_cb, cls); } /** - * Abort story telling. + * Request to replay the latest messages from the message history of the channel. * - * This function must not be called from within method handlers (as given to - * GNUNET_PSYC_slave_join()) of the slave. + * 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. + * + * @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. * - * @param story Story telling operation to stop. + * @return Handle to cancel history replay operation. */ +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 channel_history_replay (chn, 0, 0, message_limit, method_prefix, flags, + message_cb, message_part_cb, result_cb, cls); +} + + void -GNUNET_PSYC_channel_story_tell_cancel (struct GNUNET_PSYC_Story *story) +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); +} + +/** + * Retrieve the best matching channel state variable. + * + * 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 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. + */ +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; } @@ -1042,21 +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); + } @@ -1070,32 +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 */