{
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);
GNUNET_SCHEDULER_cancel (h->reconnect_task);
h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
}
- if (NULL != h->th)
- {
- GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
- h->th = NULL;
- }
while (NULL != (qe = h->queue_head))
{
GNUNET_assert (NULL != qe->response_proc);
const struct GNUNET_SCHEDULER_TaskContext *tc)
{
struct GNUNET_DATASTORE_QueueEntry *qe = cls;
- int in_receive;
GNUNET_STATISTICS_update (qe->h->stats,
gettext_noop ("# queue entry timeouts"),
GNUNET_NO);
qe->task = GNUNET_SCHEDULER_NO_TASK;
GNUNET_assert (qe->was_transmitted == GNUNET_NO);
- in_receive = qe->h->in_receive;
qe->response_proc (qe->h, NULL);
- qe->h->in_receive = in_receive;
}
struct GNUNET_DATASTORE_QueueEntry *ret;
struct GNUNET_DATASTORE_QueueEntry *pos;
unsigned int c;
- int in_receive;
c = 0;
pos = h->queue_head;
ret->task = GNUNET_SCHEDULER_add_delayed (timeout,
&timeout_queue_entry,
ret);
- in_receive = h->in_receive;
pos = ret->next;
while (pos != NULL)
{
}
pos = pos->next;
}
- h->in_receive = in_receive;
return ret;
}
}
+/**
+ * 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.
*
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,
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;