*/
struct GNUNET_MQ_Handle *mq;
+ /**
+ * Task to allow mq to send more traffic.
+ */
+ struct GNUNET_SCHEDULER_Task *mq_cont;
+
+ /**
+ * Pending envelope in case we don't have an ACK from the service.
+ */
+ struct GNUNET_MQ_Envelope *pending_env;
+
/**
* Window change handler.
*/
GNUNET_CONTAINER_DLL_remove (h->channels_head,
h->channels_tail,
ch);
-
+ if (NULL != ch->mq_cont)
+ {
+ GNUNET_SCHEDULER_cancel (ch->mq_cont);
+ ch->mq_cont = NULL;
+ }
/* signal channel destruction */
if (0 != ch->peer)
{
}
+/**
+ * Notify the application about a change in the window size (if needed).
+ *
+ * @param ch Channel to notify about.
+ */
+static void
+notify_window_size (struct GNUNET_CADET_Channel *ch)
+{
+ if (NULL != ch->window_changes)
+ {
+ ch->window_changes (ch->ctx, ch, ch->allow_send);
+ }
+}
+
/******************************************************************************/
/*********************** MQ API CALLBACKS ****************************/
/******************************************************************************/
+/**
+ * Allow the MQ implementation to send the next message.
+ *
+ * @param cls Closure (channel whose mq to activate).
+ */
+static void
+cadet_mq_send_continue (void *cls)
+{
+ struct GNUNET_CADET_Channel *ch = cls;
+
+ ch->mq_cont = NULL;
+ GNUNET_MQ_impl_send_continue (ch->mq);
+}
/**
* Implement sending functionality of a message queue for
GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA,
msg);
cadet_msg->ccn = ch->ccn;
- GNUNET_MQ_send (h->mq, env);
- GNUNET_MQ_impl_send_continue (mq);
+
+ if (0 < ch->allow_send)
+ {
+ /* Service has allowed this message, just send it and continue accepting */
+ GNUNET_MQ_send (h->mq, env);
+ ch->allow_send--;
+ ch->mq_cont = GNUNET_SCHEDULER_add_now (&cadet_mq_send_continue, ch);
+ // notify_window_size (ch); /* FIXME add "verbose" setting? */
+ }
+ else
+ {
+ /* Service has NOT allowed this message, queue it and wait for an ACK */
+ GNUNET_assert (NULL != ch->pending_env);
+ ch->pending_env = env;
+ }
}
* @param mq message queue
* @param impl_state state specific to the implementation
*/
+
static void
cadet_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
void *impl_state)
*
* @param h Cadet handle.
* @param message Message itself.
- *
- * FIXME either delete or port to MQ
*/
static void
handle_local_ack (void *cls,
return;
}
ch->allow_send++;
- LOG (GNUNET_ERROR_TYPE_DEBUG,
- "Got an ACK on channel %X, allow send now %u!\n",
- ntohl (ch->ccn.channel_of_client),
- ch->allow_send);
+ if (NULL != ch->mq)
+ {
+ if (NULL == ch->pending_env)
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got an ACK on mq channel %X, allow send now %u!\n",
+ ntohl (ch->ccn.channel_of_client),
+ ch->allow_send);
+ notify_window_size (ch);
+ }
+ else
+ {
+ LOG (GNUNET_ERROR_TYPE_DEBUG,
+ "Got an ACK on channel %X, sending pending message!\n",
+ ntohl (ch->ccn.channel_of_client));
+ GNUNET_MQ_send (h->mq, ch->pending_env);
+ ch->allow_send--;
+ ch->pending_env = NULL;
+ ch->mq_cont = GNUNET_SCHEDULER_add_now (&cadet_mq_send_continue, ch);
+ }
+ return;
+ }
+
+ /** @deprecated */
+ /* Old style API */
for (th = h->th_head; NULL != th; th = th->next)
{
if ( (th->channel == ch) &&