{
struct GNUNET_DATASTORE_QueueEntry *qe;
+ if (NULL != h->th)
+ {
+ GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
+ h->th = NULL;
+ }
if (h->client != NULL)
{
GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
h))
return;
GNUNET_CLIENT_disconnect (h->client, GNUNET_NO);
+ h->client = NULL;
}
GNUNET_break (0);
}
GNUNET_STATISTICS_destroy (h->stats,
GNUNET_NO);
+ h->stats = NULL;
GNUNET_free (h);
}
1,
GNUNET_NO);
qe->task = GNUNET_SCHEDULER_NO_TASK;
- GNUNET_assert (qe->was_transmitted == GNUNET_NO);
+ GNUNET_assert (qe->was_transmitted == GNUNET_NO);
qe->response_proc (qe->h, NULL);
}
pos = ret->next;
while (pos != NULL)
{
- if (pos->max_queue < h->queue_size)
+ if ( (pos->max_queue < h->queue_size) &&
+ (pos->was_transmitted == GNUNET_NO) )
{
GNUNET_assert (pos->response_proc != NULL);
/* move 'pos' element to head so that it will be
GNUNET_CONTAINER_DLL_insert (h->queue_head,
h->queue_tail,
pos);
+ GNUNET_STATISTICS_update (h->stats,
+ gettext_noop ("# Requests dropped from datastore queue"),
+ 1,
+ GNUNET_NO);
pos->response_proc (h, NULL);
break;
}
}
+/**
+ * Function called whenever we receive a message from
+ * the service. Calls the appropriate handler.
+ *
+ * @param cls the 'struct GNUNET_DATASTORE_Handle'
+ * @param msg the received message
+ */
+static void
+receive_cb (void *cls,
+ const struct GNUNET_MessageHeader *msg)
+{
+ struct GNUNET_DATASTORE_Handle *h = cls;
+ struct GNUNET_DATASTORE_QueueEntry *qe;
+
+ h->in_receive = GNUNET_NO;
+ if (h->skip_next_messages > 0)
+ {
+ h->skip_next_messages--;
+ process_queue (h);
+ return;
+ }
+ if (NULL == (qe = h->queue_head))
+ {
+ GNUNET_break (0);
+ process_queue (h);
+ return;
+ }
+ qe->response_proc (h, msg);
+}
+
+
/**
* Transmit request from queue to datastore service.
*
qe->was_transmitted = GNUNET_YES;
GNUNET_SCHEDULER_cancel (qe->task);
qe->task = GNUNET_SCHEDULER_NO_TASK;
+ GNUNET_assert (GNUNET_NO == h->in_receive);
h->in_receive = GNUNET_YES;
GNUNET_CLIENT_receive (h->client,
- qe->response_proc,
+ &receive_cb,
h,
GNUNET_TIME_absolute_get_remaining (qe->timeout));
GNUNET_STATISTICS_update (h->stats,
GNUNET_YES,
&transmit_request,
h);
+ GNUNET_assert (GNUNET_NO == h->in_receive);
+ GNUNET_break (NULL != h->th);
}
int32_t status;
int was_transmitted;
- h->in_receive = GNUNET_NO;
- if (h->skip_next_messages > 0)
- {
- h->skip_next_messages--;
- process_queue (h);
- return;
- }
if (NULL == (qe = h->queue_head))
{
GNUNET_break (0);
struct ResultContext rc;
const struct DataMessage *dm;
- h->in_receive = GNUNET_NO;
- if (h->skip_next_messages > 0)
- {
- h->skip_next_messages--;
- process_queue (h);
- return;
- }
if (msg == NULL)
{
qe = h->queue_head;