}
while (NULL != (qe = h->queue_head))
{
- if (NULL != qe->response_proc)
- {
- qe->response_proc (qe, NULL);
- }
- else
- {
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- if (qe->task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (h->sched,
- qe->task);
- GNUNET_free (qe);
- }
+ GNUNET_assert (NULL != qe->response_proc);
+ qe->response_proc (qe, NULL);
}
if (GNUNET_YES == drop)
{
c++;
pos = pos->next;
}
- if (c >= max_queue_size)
- return NULL;
+ ret = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_QueueEntry) + msize);
+ ret->h = h;
+ ret->response_proc = response_proc;
+ ret->client_ctx = client_ctx;
+ ret->timeout = GNUNET_TIME_relative_to_absolute (timeout);
+ ret->priority = queue_priority;
+ ret->max_queue = max_queue_size;
+ ret->message_size = msize;
+ ret->was_transmitted = GNUNET_NO;
if (pos == NULL)
{
/* append at the tail */
(h->queue_head->was_transmitted) )
pos = h->queue_head;
}
- ret = GNUNET_malloc (sizeof (struct GNUNET_DATASTORE_QueueEntry) + msize);
+ c++;
GNUNET_CONTAINER_DLL_insert_after (h->queue_head,
h->queue_tail,
pos,
ret);
- ret->h = h;
- ret->response_proc = response_proc;
- ret->client_ctx = client_ctx;
+ h->queue_size++;
+ if (c > max_queue_size)
+ {
+ response_proc (ret, NULL);
+ GNUNET_free (ret);
+ return NULL;
+ }
ret->task = GNUNET_SCHEDULER_add_delayed (h->sched,
timeout,
&timeout_queue_entry,
ret);
- ret->timeout = GNUNET_TIME_relative_to_absolute (timeout);
- ret->priority = queue_priority;
- ret->max_queue = max_queue_size;
- ret->message_size = msize;
- ret->was_transmitted = GNUNET_NO;
- h->queue_size++;
- c++;
pos = ret->next;
while (pos != NULL)
{
if (pos->max_queue < h->queue_size)
{
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- pos);
- GNUNET_SCHEDULER_cancel (h->sched,
- pos->task);
- if (pos->response_proc != NULL)
- pos->response_proc (pos, NULL);
- GNUNET_free (pos);
- h->queue_size--;
+ GNUNET_assert (pos->response_proc != NULL);
+ pos->response_proc (pos, NULL);
break;
}
pos = pos->next;
* Disconnect from the service and then try reconnecting to the datastore service
* after some delay.
*
- * @param cls the 'struct GNUNET_DATASTORE_Handle'
- * @param tc scheduler context
+ * @param h handle to datastore to disconnect and reconnect
*/
static void
do_disconnect (struct GNUNET_DATASTORE_Handle *h)
}
+static void
+free_queue_entry (struct GNUNET_DATASTORE_QueueEntry *qe)
+{
+ struct GNUNET_DATASTORE_Handle *h = qe->h;
+
+ GNUNET_CONTAINER_DLL_remove (h->queue_head,
+ h->queue_tail,
+ qe);
+ if (qe->task != GNUNET_SCHEDULER_NO_TASK)
+ {
+ GNUNET_SCHEDULER_cancel (h->sched,
+ qe->task);
+ qe->task = GNUNET_SCHEDULER_NO_TASK;
+ }
+ h->queue_size--;
+ GNUNET_free (qe);
+}
+
/**
* Type of a function to call when we receive a message
* from the service.
const char *emsg;
int32_t status;
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- if (qe->task != GNUNET_SCHEDULER_NO_TASK)
- {
- GNUNET_SCHEDULER_cancel (h->sched,
- qe->task);
- qe->task = GNUNET_SCHEDULER_NO_TASK;
- }
- GNUNET_free (qe);
+ free_queue_entry (qe);
if (msg == NULL)
{
if (NULL == h->client)
GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
_("Failed to receive response from datastore\n"));
#endif
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- GNUNET_free (qe);
+ free_queue_entry (qe);
do_disconnect (h);
rc->iter (rc->iter_cls,
NULL, 0, NULL, 0, 0, 0,
GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
"Received end of result set\n");
#endif
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- GNUNET_free (qe);
+ free_queue_entry (qe);
rc->iter (rc->iter_cls,
NULL, 0, NULL, 0, 0, 0,
GNUNET_TIME_UNIT_ZERO_ABS, 0);
(ntohs(msg->size) != sizeof(struct DataMessage) + ntohl (((const struct DataMessage*)msg)->size)) )
{
GNUNET_break (0);
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- GNUNET_free (qe);
+ free_queue_entry (qe);
h->retry_time = GNUNET_TIME_UNIT_ZERO;
do_disconnect (h);
rc->iter (rc->iter_cls,
queue_priority, max_queue_size, timeout,
&process_result_message, rcont);
if (qe == NULL)
- return NULL;
+ return NULL;
m = (struct GNUNET_MessageHeader*) &qe[1];
m->type = htons(GNUNET_MESSAGE_TYPE_DATASTORE_GET_RANDOM);
m->size = htons(sizeof (struct GNUNET_MessageHeader));
GNUNET_TIME_absolute_get_remaining (qe->timeout));
return;
}
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- GNUNET_free (qe);
+ free_queue_entry (qe);
h->retry_time = GNUNET_TIME_UNIT_ZERO;
do_disconnect (h);
rc->iter (rc->iter_cls,
h = qe->h;
reconnect = qe->was_transmitted;
- GNUNET_CONTAINER_DLL_remove (h->queue_head,
- h->queue_tail,
- qe);
- if (qe->task != GNUNET_SCHEDULER_NO_TASK)
- GNUNET_SCHEDULER_cancel (h->sched,
- qe->task);
- GNUNET_free (qe);
+ free_queue_entry (qe);
+ h->queue_size--;
if (reconnect)
{
h->retry_time = GNUNET_TIME_UNIT_ZERO;