transport_api2_communication.c

Go to the documentation of this file.

/*
     This file is part of GNUnet.
     Copyright (C) 2018 GNUnet e.V.
 
     GNUnet is free software: you can redistribute it and/or modify it
     under the terms of the GNU Affero General Public License as published
     by the Free Software Foundation, either version 3 of the License,
     or (at your option) any later version.
 
     GNUnet is distributed in the hope that it will be useful, but
     WITHOUT ANY WARRANTY; without even the implied warranty of
     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
     Affero General Public License for more details.
 
     You should have received a copy of the GNU Affero General Public License
     along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
     SPDX-License-Identifier: AGPL3.0-or-later
 */
 
#include "platform.h"
#include "gnunet_util_lib.h"
#include "gnunet_protocols.h"
#include "gnunet_transport_communication_service.h"
#include "transport.h"
 
 
#define DEFAULT_MAX_QUEUE_LENGTH 16
 
 
struct FlowControl
{
  struct FlowControl *next;
 
  struct FlowControl *prev;
 
  GNUNET_TRANSPORT_MessageCompletedCallback cb;
 
  void *cb_cls;
 
  struct GNUNET_PeerIdentity sender;
 
  uint64_t id;
};
 
 
struct AckPending
{
  struct AckPending *next;
 
  struct AckPending *prev;
 
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 
  struct GNUNET_PeerIdentity receiver;
 
  uint64_t mid;
 
  uint32_t qid;
};
 
 
struct GNUNET_TRANSPORT_CommunicatorHandle
{
  struct GNUNET_TRANSPORT_AddressIdentifier *ai_head;
 
  struct GNUNET_TRANSPORT_AddressIdentifier *ai_tail;
 
  struct FlowControl *fc_head;
 
  struct FlowControl *fc_tail;
 
  struct AckPending *ap_head;
 
  struct AckPending *ap_tail;
 
  struct GNUNET_TRANSPORT_QueueHandle *queue_head;
 
  struct GNUNET_TRANSPORT_QueueHandle *queue_tail;
 
  const struct GNUNET_CONFIGURATION_Handle *cfg;
 
  GNUNET_TRANSPORT_StartBurstNotify sb;
 
  const char *config_section;
 
  const char *addr_prefix;
 
  GNUNET_TRANSPORT_CommunicatorMqInit mq_init;
 
  void *mq_init_cls;
 
  GNUNET_TRANSPORT_CommunicatorNotify notify_cb;
 
  void *notify_cb_cls;
 
  struct GNUNET_MQ_Handle *mq;
 
  unsigned long long max_queue_length;
 
  uint64_t fc_gen;
 
  uint32_t aid_gen;
 
  uint32_t queue_gen;
 
  enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc;
};
 
 
struct GNUNET_TRANSPORT_QueueHandle
{
  struct GNUNET_TRANSPORT_QueueHandle *next;
 
  struct GNUNET_TRANSPORT_QueueHandle *prev;
 
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 
  char *address;
 
  struct GNUNET_MQ_Handle *mq;
 
  struct GNUNET_PeerIdentity peer;
 
  enum GNUNET_NetworkType nt;
 
  enum GNUNET_TRANSPORT_ConnectionStatus cs;
 
  uint32_t queue_id;
 
  uint32_t mtu;
 
  uint64_t q_len;
  uint32_t priority;
};
 
 
struct GNUNET_TRANSPORT_AddressIdentifier
{
  struct GNUNET_TRANSPORT_AddressIdentifier *next;
 
  struct GNUNET_TRANSPORT_AddressIdentifier *prev;
 
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 
  char *address;
 
  struct GNUNET_TIME_Relative expiration;
 
  uint32_t aid;
 
  enum GNUNET_NetworkType nt;
};
 
 
static void
reconnect (struct GNUNET_TRANSPORT_CommunicatorHandle *ch);
 
 
static void
send_add_address (struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_AddAddressMessage *aam;
 
  if (NULL == ai->ch->mq)
    return;
  env = GNUNET_MQ_msg_extra (aam,
                             strlen (ai->address) + 1,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS);
  aam->expiration = GNUNET_TIME_relative_hton (ai->expiration);
  aam->nt = htonl ((uint32_t) ai->nt);
  memcpy (&aam[1], ai->address, strlen (ai->address) + 1);
  GNUNET_MQ_send (ai->ch->mq, env);
}
 
 
static void
send_del_address (struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_DelAddressMessage *dam;
 
  if (NULL == ai->ch->mq)
    return;
  env = GNUNET_MQ_msg (dam, GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS);
  dam->aid = htonl (ai->aid);
  GNUNET_MQ_send (ai->ch->mq, env);
}
 
 
static void
send_add_queue (struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_AddQueueMessage *aqm;
 
  if (NULL == qh->ch->mq)
    return;
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Sending `GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP` message\n");
  env = GNUNET_MQ_msg_extra (aqm,
                             strlen (qh->address) + 1,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP);
  aqm->qid = htonl (qh->queue_id);
  aqm->receiver = qh->peer;
  aqm->nt = htonl ((uint32_t) qh->nt);
  aqm->mtu = htonl (qh->mtu);
  aqm->q_len = GNUNET_htonll (qh->q_len);
  aqm->priority = htonl (qh->priority);
  aqm->cs = htonl ((uint32_t) qh->cs);
  memcpy (&aqm[1], qh->address, strlen (qh->address) + 1);
  GNUNET_MQ_send (qh->ch->mq, env);
}
 
 
static void
send_update_queue (struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_UpdateQueueMessage *uqm;
 
  if (NULL == qh->ch->mq)
    return;
  env = GNUNET_MQ_msg (uqm, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_UPDATE);
  uqm->qid = htonl (qh->queue_id);
  uqm->receiver = qh->peer;
  uqm->nt = htonl ((uint32_t) qh->nt);
  uqm->mtu = htonl (qh->mtu);
  uqm->q_len = GNUNET_htonll (qh->q_len);
  uqm->priority = htonl (qh->priority);
  uqm->cs = htonl ((uint32_t) qh->cs);
  GNUNET_MQ_send (qh->ch->mq, env);
}
 
 
static void
send_del_queue (struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_DelQueueMessage *dqm;
 
  if (NULL == qh->ch->mq)
    return;
  env = GNUNET_MQ_msg (dqm, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN);
  dqm->qid = htonl (qh->queue_id);
  dqm->receiver = qh->peer;
  GNUNET_MQ_send (qh->ch->mq, env);
}
 
 
static void
disconnect (struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  struct FlowControl *fcn;
  struct AckPending *apn;
 
  for (struct FlowControl *fc = ch->fc_head; NULL != fc; fc = fcn)
  {
    fcn = fc->next;
    GNUNET_CONTAINER_DLL_remove (ch->fc_head, ch->fc_tail, fc);
    fc->cb (fc->cb_cls, GNUNET_SYSERR);
    GNUNET_free (fc);
  }
  for (struct AckPending *ap = ch->ap_head; NULL != ap; ap = apn)
  {
    apn = ap->next;
    GNUNET_CONTAINER_DLL_remove (ch->ap_head, ch->ap_tail, ap);
    GNUNET_free (ap);
  }
  if (NULL == ch->mq)
    return;
  GNUNET_MQ_destroy (ch->mq);
  ch->mq = NULL;
}
 
 
static void
error_handler (void *cls, enum GNUNET_MQ_Error error)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_INFO,
              "MQ failure %d, reconnecting to transport service.\n",
              error);
  disconnect (ch);
  /* TODO: maybe do this with exponential backoff/delay */
  reconnect (ch);
}
 
 
static void
handle_incoming_ack (
  void *cls,
  const struct GNUNET_TRANSPORT_IncomingMessageAck *incoming_ack)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
 
  for (struct FlowControl *fc = ch->fc_head; NULL != fc; fc = fc->next)
  {
    if ((fc->id == incoming_ack->fc_id) &&
        (0 == memcmp (&fc->sender,
                      &incoming_ack->sender,
                      sizeof(struct GNUNET_PeerIdentity))))
    {
      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                  "Done with message with flow control id %" PRIu64
                  " for sender %s from sender %s\n",
                  incoming_ack->fc_id,
                  GNUNET_i2s (&fc->sender),
                  GNUNET_i2s (&incoming_ack->sender));
      GNUNET_CONTAINER_DLL_remove (ch->fc_head, ch->fc_tail, fc);
      fc->cb (fc->cb_cls, GNUNET_OK);
      GNUNET_free (fc);
      return;
    }
  }
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Message with flow control id %" PRIu64
              " from sender %s not found\n",
              incoming_ack->fc_id,
              GNUNET_i2s (&incoming_ack->sender));
  GNUNET_break (0);
  disconnect (ch);
  /* TODO: maybe do this with exponential backoff/delay */
  reconnect (ch);
}
 
 
static int
check_create_queue (void *cls, const struct GNUNET_TRANSPORT_CreateQueue *cq)
{
  (void) cls;
  GNUNET_MQ_check_zero_termination (cq);
  return GNUNET_OK;
}
 
 
static void
handle_create_queue (void *cls, const struct GNUNET_TRANSPORT_CreateQueue *cq)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  const char *addr = (const char *) &cq[1];
  struct GNUNET_TRANSPORT_CreateQueueResponse *cqr;
  struct GNUNET_MQ_Envelope *env;
  int ret = ch->mq_init (ch->mq_init_cls, &cq->receiver, addr);
 
  if (GNUNET_NO == ret)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Address `%s' is already (being) connected to.\n",
                addr);
    env = GNUNET_MQ_msg (cqr, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL);
  }
  else if (GNUNET_SYSERR == ret)
  {
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Address `%s' invalid for this communicator\n",
                addr);
    env = GNUNET_MQ_msg (cqr, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL);
  }
  else
  {
    env = GNUNET_MQ_msg (cqr, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK);
  }
  cqr->request_id = cq->request_id;
  GNUNET_MQ_send (ch->mq, env);
}
 
 
static int
check_send_msg (void *cls, const struct GNUNET_TRANSPORT_SendMessageTo *smt)
{
  (void) cls;
  GNUNET_MQ_check_boxed_message (smt);
  return GNUNET_OK;
}
 
 
static void
send_ack (struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
          int status,
          const struct GNUNET_PeerIdentity *receiver,
          uint64_t mid,
          uint32_t qid)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_SendMessageToAck *ack;
 
  env = GNUNET_MQ_msg (ack, GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK);
  ack->status = htonl (status);
  ack->mid = mid;
  ack->qid = qid;
  ack->receiver = *receiver;
  GNUNET_MQ_send (ch->mq, env);
}
 
 
static void
send_ack_cb (void *cls)
{
  struct AckPending *ap = cls;
 
  GNUNET_assert (NULL != ap);
  GNUNET_assert (NULL != ap->ch->ap_head);
  GNUNET_CONTAINER_DLL_remove (ap->ch->ap_head, ap->ch->ap_tail, ap);
  send_ack (ap->ch, GNUNET_OK, &ap->receiver, ap->mid, ap->qid);
  GNUNET_free (ap);
}
 
 
static void
handle_send_msg (void *cls, const struct GNUNET_TRANSPORT_SendMessageTo *smt)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  const struct GNUNET_MessageHeader *mh;
  struct GNUNET_MQ_Envelope *env;
  struct AckPending *ap;
  struct GNUNET_TRANSPORT_QueueHandle *qh;
 
  for (qh = ch->queue_head; NULL != qh; qh = qh->next)
    if ((qh->queue_id == ntohl (smt->qid)) &&
        (0 == memcmp (&qh->peer,
                      &smt->receiver,
                      sizeof(struct GNUNET_PeerIdentity))))
      break;
  if (NULL == qh)
  {
    /* queue is already gone, tell transport this one failed */
    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
                "Transmission failed, queue no longer exists.\n");
    send_ack (ch, GNUNET_NO, &smt->receiver, smt->mid, smt->qid);
    return;
  }
  ap = GNUNET_new (struct AckPending);
  ap->ch = ch;
  ap->receiver = smt->receiver;
  ap->mid = smt->mid;
  ap->qid = smt->qid;
  GNUNET_CONTAINER_DLL_insert (ch->ap_head, ch->ap_tail, ap);
  mh = (const struct GNUNET_MessageHeader *) &smt[1];
  env = GNUNET_MQ_msg_copy (mh);
  GNUNET_MQ_notify_sent (env, &send_ack_cb, ap);
  GNUNET_MQ_send (qh->mq, env);
}
 
 
static int
check_backchannel_incoming (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *bi)
{
  (void) cls;
  GNUNET_MQ_check_boxed_message (bi);
  return GNUNET_OK;
}
 
 
static void
handle_backchannel_incoming (
  void *cls,
  const struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming *bi)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  if (NULL != ch->notify_cb)
    ch->notify_cb (ch->notify_cb_cls,
                   &bi->pid,
                   (const struct GNUNET_MessageHeader *) &bi[1]);
  else
    GNUNET_log (
      GNUNET_ERROR_TYPE_INFO,
      _ ("Dropped backchanel message: handler not provided by communicator\n"));
}
 
 
static int
check_start_burst (void *cls,
                   const struct GNUNET_TRANSPORT_StartBurst *sb)
{
  const char *addr_str = (const char *) &sb[1];
  (void) cls;
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "check_start_burst %s %lu\n",
              addr_str,
              strlen (addr_str));
  GNUNET_MQ_check_zero_termination (sb);
  return GNUNET_OK;
}
 
 
static void
handle_start_burst (void *cls,
                    const struct GNUNET_TRANSPORT_StartBurst *sb)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = cls;
  const char *addr = (const char *) &sb[1];
  struct GNUNET_TIME_Relative rtt = GNUNET_TIME_relative_ntoh (sb->rtt);
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "Calling communicator to start burst to %s is %s rtt %lu\n",
              addr,
              NULL == sb ? "not possible" : "possible",
              (unsigned long) rtt.rel_value_us);
 
  if (NULL != ch->sb)
    ch->sb (addr, GNUNET_TIME_relative_ntoh (sb->rtt), (struct
                                                        GNUNET_PeerIdentity *) &
            sb->pid);
}
 
 
static void
reconnect (struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  struct GNUNET_MQ_MessageHandler handlers[] =
  { GNUNET_MQ_hd_fixed_size (incoming_ack,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG_ACK,
                             struct GNUNET_TRANSPORT_IncomingMessageAck,
                             ch),
    GNUNET_MQ_hd_var_size (create_queue,
                           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE,
                           struct GNUNET_TRANSPORT_CreateQueue,
                           ch),
    GNUNET_MQ_hd_var_size (send_msg,
                           GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG,
                           struct GNUNET_TRANSPORT_SendMessageTo,
                           ch),
    GNUNET_MQ_hd_var_size (
      backchannel_incoming,
      GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL_INCOMING,
      struct GNUNET_TRANSPORT_CommunicatorBackchannelIncoming,
      ch),
    GNUNET_MQ_hd_var_size (
      start_burst,
      GNUNET_MESSAGE_TYPE_TRANSPORT_START_BURST,
      struct GNUNET_TRANSPORT_StartBurst,
      ch),
    GNUNET_MQ_handler_end () };
  struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *cam;
  struct GNUNET_MQ_Envelope *env;
 
  ch->mq =
    GNUNET_CLIENT_connect (ch->cfg, "transport", handlers, &error_handler, ch);
  if (NULL == ch->mq)
    return;
  env = GNUNET_MQ_msg_extra (cam,
                             strlen (ch->addr_prefix) + 1,
                             GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR);
  if  (NULL != ch->sb)
  {
    cam->can_burst = htonl (GNUNET_YES);
  }
  cam->cc = htonl ((uint32_t) ch->cc);
  memcpy (&cam[1], ch->addr_prefix, strlen (ch->addr_prefix) + 1);
  GNUNET_MQ_send (ch->mq, env);
  for (struct GNUNET_TRANSPORT_AddressIdentifier *ai = ch->ai_head; NULL != ai;
       ai = ai->next)
    send_add_address (ai);
  for (struct GNUNET_TRANSPORT_QueueHandle *qh = ch->queue_head; NULL != qh;
       qh = qh->next)
    send_add_queue (qh);
}
 
 
struct GNUNET_TRANSPORT_CommunicatorHandle *
GNUNET_TRANSPORT_communicator_connect (
  const struct GNUNET_CONFIGURATION_Handle *cfg,
  const char *config_section,
  const char *addr_prefix,
  enum GNUNET_TRANSPORT_CommunicatorCharacteristics cc,
  GNUNET_TRANSPORT_CommunicatorMqInit mq_init,
  void *mq_init_cls,
  GNUNET_TRANSPORT_CommunicatorNotify notify_cb,
  void *notify_cb_cls,
  GNUNET_TRANSPORT_StartBurstNotify sb)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch;
 
  ch = GNUNET_new (struct GNUNET_TRANSPORT_CommunicatorHandle);
  ch->cfg = cfg;
  ch->config_section = config_section;
  ch->addr_prefix = addr_prefix;
  ch->mq_init = mq_init;
  ch->mq_init_cls = mq_init_cls;
  ch->notify_cb = notify_cb;
  ch->notify_cb_cls = notify_cb_cls;
  ch->cc = cc;
  ch->sb = sb;
  reconnect (ch);
  if (GNUNET_OK !=
      GNUNET_CONFIGURATION_get_value_number (cfg,
                                             config_section,
                                             "MAX_QUEUE_LENGTH",
                                             &ch->max_queue_length))
    ch->max_queue_length = DEFAULT_MAX_QUEUE_LENGTH;
  if (NULL == ch->mq)
  {
    GNUNET_free (ch);
    return NULL;
  }
  return ch;
}
 
 
void
GNUNET_TRANSPORT_communicator_disconnect (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  disconnect (ch);
  while (NULL != ch->ai_head)
  {
    GNUNET_break (0);  /* communicator forgot to remove address, warn! */
    GNUNET_TRANSPORT_communicator_address_remove (ch->ai_head);
  }
  GNUNET_free (ch);
}
 
 
/* ************************* Receiving *************************** */
 
 
int
GNUNET_TRANSPORT_communicator_receive (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *sender,
  const struct GNUNET_MessageHeader *msg,
  struct GNUNET_TIME_Relative expected_addr_validity,
  GNUNET_TRANSPORT_MessageCompletedCallback cb,
  void *cb_cls)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_IncomingMessage *im;
  uint16_t msize;
 
 
  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
              "communicator receive\n");
 
  if (NULL == ch->mq)
    return GNUNET_SYSERR;
  if ((NULL == cb) && (GNUNET_MQ_get_length (ch->mq) >= ch->max_queue_length))
  {
    GNUNET_log (
      GNUNET_ERROR_TYPE_WARNING,
      "Dropping message: transport is too slow, queue length %llu exceeded\n",
      ch->max_queue_length);
    return GNUNET_NO;
  }
 
  msize = ntohs (msg->size);
  env =
    GNUNET_MQ_msg_extra (im, msize, GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG);
  if (NULL == env)
  {
    GNUNET_break (0);
    return GNUNET_SYSERR;
  }
  im->expected_address_validity =
    GNUNET_TIME_relative_hton (expected_addr_validity);
  im->sender = *sender;
  // FIXME: this is expensive, would be better if we would
  // re-design the API to allow us to create the envelope first,
  // and then have the application fill in the body so we do
  // not have to memcpy()
  memcpy (&im[1], msg, msize);
  im->fc_on = htonl (GNUNET_NO);
  if (NULL != cb)
  {
    struct FlowControl *fc;
 
    im->fc_on = htonl (GNUNET_YES);
    im->fc_id = ch->fc_gen++;
    fc = GNUNET_new (struct FlowControl);
    fc->sender = *sender;
    fc->id = im->fc_id;
    fc->cb = cb;
    fc->cb_cls = cb_cls;
    GNUNET_CONTAINER_DLL_insert (ch->fc_head, ch->fc_tail, fc);
    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
                "Created flow control id %" PRIu64 " for sender %s\n",
                fc->id,
                GNUNET_i2s (&fc->sender));
  }
  GNUNET_MQ_send (ch->mq, env);
  return GNUNET_OK;
}
 
 
/* ************************* Discovery *************************** */
 
 
struct GNUNET_TRANSPORT_QueueHandle *
GNUNET_TRANSPORT_communicator_mq_add (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *peer,
  const char *address,
  uint32_t mtu,
  uint64_t q_len,
  uint32_t priority,
  enum GNUNET_NetworkType nt,
  enum GNUNET_TRANSPORT_ConnectionStatus cs,
  struct GNUNET_MQ_Handle *mq)
{
  struct GNUNET_TRANSPORT_QueueHandle *qh;
 
  // Do not notify the service if there is no initial capacity.
  GNUNET_assert (0 < q_len);
 
  qh = GNUNET_new (struct GNUNET_TRANSPORT_QueueHandle);
  qh->ch = ch;
  qh->peer = *peer;
  qh->address = GNUNET_strdup (address);
  qh->nt = nt;
  qh->mtu = mtu;
  qh->q_len = q_len;
  qh->priority = priority;
  qh->cs = cs;
  qh->mq = mq;
  qh->queue_id = ch->queue_gen++;
  GNUNET_CONTAINER_DLL_insert (ch->queue_head, ch->queue_tail, qh);
  send_add_queue (qh);
  return qh;
}
 
 
void
GNUNET_TRANSPORT_communicator_mq_update (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_TRANSPORT_QueueHandle *u_qh,
  uint64_t q_len,
  uint32_t priority)
{
  struct GNUNET_TRANSPORT_QueueHandle *qh;
 
  for (qh = ch->queue_head; NULL != qh; qh = qh->next)
  {
    if (u_qh == qh)
      break;
  }
  GNUNET_assert (NULL != qh);
  qh->q_len = q_len;
  qh->priority = priority;
  send_update_queue (qh);
}
 
 
void
GNUNET_TRANSPORT_communicator_mq_del (struct GNUNET_TRANSPORT_QueueHandle *qh)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = qh->ch;
 
  send_del_queue (qh);
  GNUNET_CONTAINER_DLL_remove (ch->queue_head, ch->queue_tail, qh);
  GNUNET_MQ_destroy (qh->mq);
  GNUNET_free (qh->address);
  GNUNET_free (qh);
}
 
 
struct GNUNET_TRANSPORT_AddressIdentifier *
GNUNET_TRANSPORT_communicator_address_add (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const char *address,
  enum GNUNET_NetworkType nt,
  struct GNUNET_TIME_Relative expiration)
{
  struct GNUNET_TRANSPORT_AddressIdentifier *ai;
 
  ai = GNUNET_new (struct GNUNET_TRANSPORT_AddressIdentifier);
  ai->ch = ch;
  ai->address = GNUNET_strdup (address);
  ai->nt = nt;
  ai->expiration = expiration;
  ai->aid = ch->aid_gen++;
  GNUNET_CONTAINER_DLL_insert (ch->ai_head, ch->ai_tail, ai);
  send_add_address (ai);
  return ai;
}
 
 
void
GNUNET_TRANSPORT_communicator_address_remove (
  struct GNUNET_TRANSPORT_AddressIdentifier *ai)
{
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch = ai->ch;
 
  send_del_address (ai);
  GNUNET_CONTAINER_DLL_remove (ch->ai_head, ch->ai_tail, ai);
  GNUNET_free (ai->address);
  GNUNET_free (ai);
  ai = NULL;
}
 
 
void
GNUNET_TRANSPORT_communicator_address_remove_all (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch)
{
  struct GNUNET_TRANSPORT_AddressIdentifier *ai = ch->ai_head;
  while (NULL != ai)
  {
    struct GNUNET_TRANSPORT_AddressIdentifier *ai_next = ai->next;
    GNUNET_TRANSPORT_communicator_address_remove (ai);
    ai = ai_next;
  }
}
 
 
/* ************************* Backchannel *************************** */
 
 
void
GNUNET_TRANSPORT_communicator_notify (
  struct GNUNET_TRANSPORT_CommunicatorHandle *ch,
  const struct GNUNET_PeerIdentity *pid,
  const char *comm,
  const struct GNUNET_MessageHeader *header)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_CommunicatorBackchannel *cb;
  size_t slen = strlen (comm) + 1;
  uint16_t mlen = ntohs (header->size);
 
  GNUNET_assert (mlen + slen + sizeof(*cb) < UINT16_MAX);
  env =
    GNUNET_MQ_msg_extra (cb,
                         slen + mlen,
                         GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL)
  ;
  cb->pid = *pid;
  memcpy (&cb[1], header, mlen);
  memcpy (((char *) &cb[1]) + mlen, comm, slen);
  GNUNET_MQ_send (ch->mq, env);
}
 
 
/* ************************* Burst *************************** */
 
void
GNUNET_TRANSPORT_communicator_burst_finished (struct
                                              GNUNET_TRANSPORT_CommunicatorHandle
                                              *ch)
{
  struct GNUNET_MQ_Envelope *env;
  struct GNUNET_TRANSPORT_BurstFinished *bf;
 
  env = GNUNET_MQ_msg (bf, GNUNET_MESSAGE_TYPE_TRANSPORT_BURST_FINISHED);
  GNUNET_MQ_send (ch->mq, env);
}
 
 
/* end of transport_api2_communication.c */