};
+/**
+ * Handle to a monitoring request.
+ */
+struct GNUNET_DHT_MonitorHandle
+{
+ /**
+ * DLL.
+ */
+ struct GNUNET_DHT_MonitorHandle *next;
+
+ /**
+ * DLL.
+ */
+ struct GNUNET_DHT_MonitorHandle *prev;
+
+ /**
+ * Main handle to this DHT api.
+ */
+ struct GNUNET_DHT_Handle *dht_handle;
+
+ /**
+ * Type of block looked for.
+ */
+ enum GNUNET_BLOCK_Type type;
+
+ /**
+ * Key being looked for, NULL == all.
+ */
+ GNUNET_HashCode *key;
+
+ /**
+ * Callback for each received message of interest.
+ */
+ GNUNET_DHT_MonitorCB cb;
+
+ /**
+ * Closure for cb.
+ */
+ void *cb_cls;
+
+};
+
+
/**
* Connection to the DHT service.
*/
*/
struct PendingMessage *pending_tail;
+ /**
+ * Head of linked list of messages we would like to monitor.
+ */
+ struct GNUNET_DHT_MonitorHandle *monitor_head;
+
+ /**
+ * Tail of linked list of messages we would like to monitor.
+ */
+ struct GNUNET_DHT_MonitorHandle *monitor_tail;
+
/**
* Hash map containing the current outstanding unique requests
* (values are of type 'struct GNUNET_DHT_RouteHandle').
struct GNUNET_DHT_GetHandle *rh = value;
if (GNUNET_NO == rh->message->in_pending_queue)
- {
+ {
+#if DEBUG_DHT
LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Retransmitting request related to %s to DHT %p\n",
- GNUNET_h2s (key),
- handle);
+ "Retransmitting request related to %s to DHT %p\n", GNUNET_h2s (key),
+ handle);
+#endif
GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
rh->message);
rh->message->in_pending_queue = GNUNET_YES;
{
struct GNUNET_DHT_Handle *handle = cls;
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Reconnecting with DHT %p\n",
- handle);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Reconnecting with DHT %p\n", handle);
+#endif
handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
if (handle->retry_time.rel_value < GNUNET_CONSTANTS_SERVICE_RETRY.rel_value)
handle->retry_time = GNUNET_CONSTANTS_SERVICE_RETRY;
handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
if (GNUNET_YES != try_connect (handle))
{
+#if DEBUG_DHT
LOG (GNUNET_ERROR_TYPE_DEBUG, "dht reconnect failed(!)\n");
+#endif
return;
}
GNUNET_CONTAINER_multihashmap_iterate (handle->active_requests,
GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
handle->th = NULL;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "Disconnecting from DHT service, will try to reconnect in %llu ms\n",
- (unsigned long long) handle->retry_time.rel_value);
+ "Disconnecting from DHT service, will try to reconnect in %llu ms\n",
+ (unsigned long long) handle->retry_time.rel_value);
GNUNET_CLIENT_disconnect (handle->client, GNUNET_NO);
handle->client = NULL;
handle->reconnect_task =
if (handle->client == NULL)
{
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "process_pending_messages called, but client is null, reconnecting\n");
+ "process_pending_messages called, but client is null, reconnecting\n");
do_disconnect (handle);
return;
}
if (NULL != handle->th)
return;
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
- "notify_transmit_ready returned NULL, reconnecting\n");
+ "notify_transmit_ready returned NULL, reconnecting\n");
do_disconnect (handle);
}
handle->th = NULL;
if (buf == NULL)
{
+#if DEBUG_DHT
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Transmission to DHT service failed! Reconnecting!\n");
+#endif
do_disconnect (handle);
return 0;
}
#endif
handle->in_receive = GNUNET_YES;
GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
- GNUNET_TIME_UNIT_FOREVER_REL);
+ GNUNET_TIME_UNIT_FOREVER_REL);
}
return tsize;
}
if (dht_msg->unique_id != get_handle->unique_id)
{
/* UID mismatch */
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Ignoring reply for %s: UID mismatch: %llu/%llu\n",
- GNUNET_h2s (key),
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Ignoring reply for %s: UID mismatch: %llu/%llu\n", GNUNET_h2s (key),
dht_msg->unique_id, get_handle->unique_id);
+#endif
return GNUNET_YES;
}
msize = ntohs (dht_msg->header.size);
return GNUNET_NO;
}
data_length = msize - meta_length;
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Giving %u byte reply for %s to application\n",
- (unsigned int) data_length,
- GNUNET_h2s (key));
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Giving %u byte reply for %s to application\n",
+ (unsigned int) data_length, GNUNET_h2s (key));
+#endif
put_path = (const struct GNUNET_PeerIdentity *) &dht_msg[1];
get_path = &put_path[put_path_length];
data = &get_path[get_path_length];
}
+/**
+ * Process a monitoring message from the service.
+ *
+ * @param handle The DHT handle.
+ * @param msg Message from the service.
+ *
+ * @return GNUNET_OK if everything went fine,
+ * GNUNET_SYSERR if the message is malformed.
+ */
+static int
+process_monitor_message (struct GNUNET_DHT_Handle *handle,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct GNUNET_DHT_MonitorMessage *m;
+ struct GNUNET_DHT_MonitorHandle *h;
+ size_t msize;
+
+ if (ntohs (msg->type) < GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET ||
+ ntohs (msg->type) > GNUNET_MESSAGE_TYPE_DHT_MONITOR_PUT)
+ return GNUNET_SYSERR;
+ msize = ntohs (msg->size);
+ if (msize < sizeof (struct GNUNET_DHT_MonitorMessage))
+ return GNUNET_SYSERR;
+
+ m = (struct GNUNET_DHT_MonitorMessage *) msg;
+ h = handle->monitor_head;
+ while (NULL != h)
+ {
+ if (h->type == ntohl(m->type) &&
+ (NULL == h->key ||
+ memcmp (h->key, &m->key, sizeof (GNUNET_HashCode)) == 0))
+ {
+ struct GNUNET_PeerIdentity *path;
+ uint32_t getl;
+ uint32_t putl;
+
+ path = (struct GNUNET_PeerIdentity *) &m[1];
+ getl = ntohl (m->get_path_length);
+ putl = ntohl (m->put_path_length);
+ h->cb (h->cb_cls, ntohs(msg->type),
+ GNUNET_TIME_absolute_ntoh(m->expiration),
+ &m->key,
+ &path[getl], putl, path, getl,
+ ntohl (m->desired_replication_level),
+ ntohl (m->options), ntohl (m->type),
+ (void *) &path[getl + putl],
+ ntohs (msg->size) -
+ sizeof (struct GNUNET_DHT_MonitorMessage) -
+ sizeof (struct GNUNET_PeerIdentity) * (putl + getl));
+ }
+ h = h->next;
+ }
+
+ return GNUNET_OK;
+}
+
/**
* Handler for messages received from the DHT service
* a demultiplexer which handles numerous message types
if (msg == NULL)
{
+#if DEBUG_DHT
LOG (GNUNET_ERROR_TYPE_DEBUG,
"Error receiving data from DHT service, reconnecting\n");
+#endif
do_disconnect (handle);
return;
}
if (ntohs (msg->type) != GNUNET_MESSAGE_TYPE_DHT_CLIENT_RESULT)
{
+ if (process_monitor_message (handle, msg) == GNUNET_OK)
+ {
+ GNUNET_CLIENT_receive (handle->client, &service_message_handler, handle,
+ GNUNET_TIME_UNIT_FOREVER_REL);
+ return;
+ }
GNUNET_break (0);
do_disconnect (handle);
return;
return;
}
dht_msg = (const struct GNUNET_DHT_ClientResultMessage *) msg;
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Received reply for `%s' from DHT service %p\n",
- GNUNET_h2s (&dht_msg->key),
- handle);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Received reply for `%s' from DHT service %p\n",
+ GNUNET_h2s (&dht_msg->key), handle);
+#endif
GNUNET_CONTAINER_multihashmap_get_multiple (handle->active_requests,
&dht_msg->key, &process_reply,
(void *) dht_msg);
struct GNUNET_DHT_Handle *handle;
handle = pending->handle;
- GNUNET_assert (GNUNET_YES == pending->in_pending_queue);
+ GNUNET_assert (GNUNET_YES == pending->in_pending_queue);
GNUNET_CONTAINER_DLL_remove (handle->pending_head, handle->pending_tail,
pending);
pending->in_pending_queue = GNUNET_NO;
GNUNET_break (0);
return NULL;
}
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sending query for %s to DHT %p\n",
- GNUNET_h2s (key),
- handle);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending query for %s to DHT %p\n",
+ GNUNET_h2s (key), handle);
+#endif
pending = GNUNET_malloc (sizeof (struct PendingMessage) + msize);
get_msg = (struct GNUNET_DHT_ClientGetMessage *) &pending[1];
pending->msg = &get_msg->header;
handle = get_handle->message->handle;
get_msg =
(const struct GNUNET_DHT_ClientGetMessage *) get_handle->message->msg;
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Sending STOP for %s to DHT via %p\n",
- GNUNET_h2s (&get_msg->key),
- handle);
+#if DEBUG_DHT
+ LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending STOP for %s to DHT via %p\n",
+ GNUNET_h2s (&get_msg->key), handle);
+#endif
/* generate STOP */
pending =
GNUNET_malloc (sizeof (struct PendingMessage) +
}
+/**
+ * Start monitoring the local DHT service.
+ *
+ * @param handle Handle to the DHT service.
+ * @param type Type of blocks that are of interest.
+ * @param key Key of data of interest, NULL for all.
+ * @param cb Callback to process all monitored data.
+ * @param cb_cls Closure for cb.
+ *
+ * @return Handle to stop monitoring.
+ */
+struct GNUNET_DHT_MonitorHandle *
+GNUNET_DHT_monitor_start (struct GNUNET_DHT_Handle *handle,
+ enum GNUNET_BLOCK_Type type,
+ const GNUNET_HashCode *key,
+ GNUNET_DHT_MonitorCB cb,
+ void *cb_cls)
+{
+ struct GNUNET_DHT_MonitorHandle *h;
+ struct GNUNET_DHT_MonitorMessage *m;
+ struct PendingMessage *pending;
+
+ h = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorHandle));
+ GNUNET_CONTAINER_DLL_insert(handle->monitor_head, handle->monitor_tail, h);
+
+ GNUNET_assert (NULL != cb);
+ h->cb = cb;
+ h->cb_cls = cb_cls;
+ h->type = type;
+ h->dht_handle = handle;
+ if (NULL != key)
+ {
+ h->key = GNUNET_malloc (sizeof(GNUNET_HashCode));
+ memcpy (h->key, key, sizeof(GNUNET_HashCode));
+ }
+
+ pending = GNUNET_malloc (sizeof (struct GNUNET_DHT_MonitorMessage) +
+ sizeof (struct PendingMessage));
+ m = (struct GNUNET_DHT_MonitorMessage *) &pending[1];
+ pending->msg = &m->header;
+ pending->handle = handle;
+ pending->free_on_send = GNUNET_YES;
+ m->header.type = htons (GNUNET_MESSAGE_TYPE_DHT_MONITOR_GET);
+ m->header.size = htons (sizeof (struct GNUNET_DHT_MonitorMessage));
+ m->type = htonl(type);
+ if (NULL != key)
+ memcpy (&m->key, key, sizeof(GNUNET_HashCode));
+ GNUNET_CONTAINER_DLL_insert (handle->pending_head, handle->pending_tail,
+ pending);
+ pending->in_pending_queue = GNUNET_YES;
+ process_pending_messages (handle);
+
+ return h;
+}
+
+
+/**
+ * Stop monitoring.
+ *
+ * @param handle The handle to the monitor request returned by monitor_start.
+ *
+ * On return get_handle will no longer be valid, caller must not use again!!!
+ */
+void
+GNUNET_DHT_monitor_stop (struct GNUNET_DHT_MonitorHandle *handle)
+{
+ GNUNET_free_non_null (handle->key);
+ GNUNET_CONTAINER_DLL_remove (handle->dht_handle->monitor_head,
+ handle->dht_handle->monitor_tail,
+ handle);
+ GNUNET_free (handle);
+}
+
+
+
/* end of dht_api.c */
projects / oweals / gnunet.git / blobdiff