*/
struct GNUNET_PeerIdentity target;
+ /**
+ * UID of our message, if any.
+ */
+ uint32_t uid;
+
};
*/
GNUNET_DV_ConnectCallback connect_cb;
+ /**
+ * Function to call on distance change events.
+ */
+ GNUNET_DV_DistanceChangedCallback distance_cb;
+
/**
* Function to call on disconnect events.
*/
th);
memcpy (&cbuf[ret], th->msg, tsize);
ret += tsize;
- (void) GNUNET_CONTAINER_multihashmap_put (sh->send_callbacks,
- &th->target.hashPubKey,
- th,
- GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ if (NULL != th->cb)
+ {
+ (void) GNUNET_CONTAINER_multihashmap_put (sh->send_callbacks,
+ &th->target.hashPubKey,
+ th,
+ GNUNET_CONTAINER_MULTIHASHMAPOPTION_MULTIPLE);
+ }
+ else
+ {
+ GNUNET_free (th);
+ }
}
return ret;
}
}
+/**
+ * Closure for 'process_ack'.
+ */
+struct AckContext
+{
+ /**
+ * The ACK message.
+ */
+ const struct GNUNET_DV_AckMessage *ack;
+
+ /**
+ * Our service handle.
+ */
+ struct GNUNET_DV_ServiceHandle *sh;
+};
+
+
+/**
+ * We got an ACK. Check if it matches the given transmit handle, and if
+ * so call the continuation.
+ *
+ * @param cls the 'struct AckContext'
+ * @param key peer identity
+ * @param value the 'struct GNUNET_DV_TransmitHandle'
+ * @return GNUNET_OK if the ACK did not match (continue to iterate)
+ */
+static int
+process_ack (void *cls,
+ const struct GNUNET_HashCode *key,
+ void *value)
+{
+ struct AckContext *ctx = cls;
+ struct GNUNET_DV_TransmitHandle *th = value;
+
+ if (th->uid != ntohl (ctx->ack->uid))
+ return GNUNET_OK;
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Matchedk ACK for message to peer %s\n",
+ GNUNET_h2s (key));
+ GNUNET_assert (GNUNET_YES ==
+ GNUNET_CONTAINER_multihashmap_remove (ctx->sh->send_callbacks,
+ key,
+ th));
+ th->cb (th->cb_cls,
+ (ntohs (ctx->ack->header.type) == GNUNET_MESSAGE_TYPE_DV_SEND_ACK)
+ ? GNUNET_OK
+ : GNUNET_SYSERR);
+ GNUNET_free (th);
+ return GNUNET_NO;
+}
+
+
/**
* Handles a message sent from the DV service to us.
* Parse it out and give it to the plugin.
{
struct GNUNET_DV_ServiceHandle *sh = cls;
const struct GNUNET_DV_ConnectMessage *cm;
+ const struct GNUNET_DV_DistanceUpdateMessage *dum;
const struct GNUNET_DV_DisconnectMessage *dm;
const struct GNUNET_DV_ReceivedMessage *rm;
const struct GNUNET_MessageHeader *payload;
-
+ const struct GNUNET_DV_AckMessage *ack;
+ struct AckContext ctx;
+
if (NULL == msg)
{
/* Connection closed */
reconnect (sh);
return;
}
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Received message of type %u from DV service\n",
+ (unsigned int) msg->type);
switch (ntohs (msg->type))
{
case GNUNET_MESSAGE_TYPE_DV_CONNECT:
&cm->peer,
ntohl (cm->distance));
break;
+ case GNUNET_MESSAGE_TYPE_DV_DISTANCE_CHANGED:
+ if (ntohs (msg->size) != sizeof (struct GNUNET_DV_DistanceUpdateMessage))
+ {
+ GNUNET_break (0);
+ reconnect (sh);
+ return;
+ }
+ dum = (const struct GNUNET_DV_DistanceUpdateMessage *) msg;
+ sh->distance_cb (sh->cls,
+ &dum->peer,
+ ntohl (dum->distance));
+ break;
case GNUNET_MESSAGE_TYPE_DV_DISCONNECT:
if (ntohs (msg->size) != sizeof (struct GNUNET_DV_DisconnectMessage))
{
ntohl (rm->distance),
payload);
break;
+ case GNUNET_MESSAGE_TYPE_DV_SEND_ACK:
+ case GNUNET_MESSAGE_TYPE_DV_SEND_NACK:
+ if (ntohs (msg->size) != sizeof (struct GNUNET_DV_AckMessage))
+ {
+ GNUNET_break (0);
+ reconnect (sh);
+ return;
+ }
+ ack = (const struct GNUNET_DV_AckMessage *) msg;
+ ctx.ack = ack;
+ ctx.sh = sh;
+ GNUNET_CONTAINER_multihashmap_get_multiple (sh->send_callbacks,
+ &ack->target.hashPubKey,
+ &process_ack,
+ &ctx);
+ break;
default:
reconnect (sh);
break;
GNUNET_CONTAINER_multihashmap_iterate (sh->send_callbacks,
&cleanup_send_cb,
sh);
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Connecting to DV service\n");
sh->client = GNUNET_CLIENT_connect ("dv", sh->cfg);
if (NULL == sh->client)
{
* @param cfg configuration
* @param cls closure for callbacks
* @param connect_cb function to call on connects
+ * @param distance_cb function to call if distances change
* @param disconnect_cb function to call on disconnects
* @param message_cb function to call if we receive messages
* @return handle to access the service
GNUNET_DV_service_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
void *cls,
GNUNET_DV_ConnectCallback connect_cb,
+ GNUNET_DV_DistanceChangedCallback distance_cb,
GNUNET_DV_DisconnectCallback disconnect_cb,
GNUNET_DV_MessageReceivedCallback message_cb)
{
sh->cfg = cfg;
sh->cls = cls;
sh->connect_cb = connect_cb;
+ sh->distance_cb = distance_cb;
sh->disconnect_cb = disconnect_cb;
sh->message_cb = message_cb;
sh->send_callbacks = GNUNET_CONTAINER_multihashmap_create (128, GNUNET_YES);
GNUNET_break (0);
return NULL;
}
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Asked to send %u bytes of type %u to %s\n",
+ (unsigned int) msg->size,
+ (unsigned int) msg->type,
+ GNUNET_i2s (target));
+
th = GNUNET_malloc (sizeof (struct GNUNET_DV_TransmitHandle) +
sizeof (struct GNUNET_DV_SendMessage) +
ntohs (msg->size));
sm->header.type = htons (GNUNET_MESSAGE_TYPE_DV_SEND);
sm->header.size = htons (sizeof (struct GNUNET_DV_SendMessage) +
ntohs (msg->size));
+ if (0 == sh->uid_gen)
+ sh->uid_gen = 1;
+ th->uid = sh->uid_gen;
+ sm->uid = htonl (sh->uid_gen++);
/* use memcpy here as 'target' may not be sufficiently aligned */
memcpy (&sm->target, target, sizeof (struct GNUNET_PeerIdentity));
memcpy (&sm[1], msg, ntohs (msg->size));