src/mesh/mesh_api.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2011 Christian Grothoff (and other contributing authors)
   4      GNUnet is free software; you can redistribute it and/or modify
   5      it under the terms of the GNU General Public License as published
   6      by the Free Software Foundation; either version 3, or (at your
   7      option) any later version.
   8      GNUnet is distributed in the hope that it will be useful, but
   9      WITHOUT ANY WARRANTY; without even the implied warranty of
  10      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  11      General Public License for more details.
  12      You should have received a copy of the GNU General Public License
  13      along with GNUnet; see the file COPYING.  If not, write to the
  14      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  15      Boston, MA 02111-1307, USA.
  16 */
  17
  18 /**
  19  * @file mesh/mesh_api.c
  20  * @brief mesh api: client implementation of new mesh service
  21  * @author Bartlomiej Polot
  22  */
  23
  24 #include "platform.h"
  25 #include "gnunet_util_lib.h"
  26 #include "gnunet_mesh_service.h"
  27 #include "mesh.h"
  28 #include "mesh_protocol.h"
  29
  30 #define LOG(kind,...) GNUNET_log_from (kind, "mesh-api",__VA_ARGS__)
  31
  32 /******************************************************************************/
  33 /************************      DATA STRUCTURES     ****************************/
  34 /******************************************************************************/
  35
  36 /**
  37  * Transmission queue to the service
  38  */
  39 struct GNUNET_MESH_TransmitHandle
  40 {
  41
  42     /**
  43      * Double Linked list
  44      */
  45   struct GNUNET_MESH_TransmitHandle *next;
  46
  47     /**
  48      * Double Linked list
  49      */
  50   struct GNUNET_MESH_TransmitHandle *prev;
  51
  52     /**
  53      * Channel this message is sent on / for (may be NULL for control messages).
  54      */
  55   struct GNUNET_MESH_Channel *channel;
  56
  57     /**
  58      * Callback to obtain the message to transmit, or NULL if we
  59      * got the message in 'data'.  Notice that messages built
  60      * by 'notify' need to be encapsulated with information about
  61      * the 'target'.
  62      */
  63   GNUNET_CONNECTION_TransmitReadyNotify notify;
  64
  65     /**
  66      * Closure for 'notify'
  67      */
  68   void *notify_cls;
  69
  70     /**
  71      * How long is this message valid.  Once the timeout has been
  72      * reached, the message must no longer be sent.  If this
  73      * is a message with a 'notify' callback set, the 'notify'
  74      * function should be called with 'buf' NULL and size 0.
  75      */
  76   struct GNUNET_TIME_Absolute timeout;
  77
  78     /**
  79      * Task triggering a timeout, can be NO_TASK if the timeout is FOREVER.
  80      */
  81   GNUNET_SCHEDULER_TaskIdentifier timeout_task;
  82
  83     /**
  84      * Size of 'data' -- or the desired size of 'notify' if 'data' is NULL.
  85      */
  86   size_t size;
  87 };
  88
  89
  90 /**
  91  * Opaque handle to the service.
  92  */
  93 struct GNUNET_MESH_Handle
  94 {
  95
  96     /**
  97      * Handle to the server connection, to send messages later
  98      */
  99   struct GNUNET_CLIENT_Connection *client;
 100
 101     /**
 102      * Set of handlers used for processing incoming messages in the channels
 103      */
 104   const struct GNUNET_MESH_MessageHandler *message_handlers;
 105
 106   /**
 107    * Number of handlers in the handlers array.
 108    */
 109   unsigned int n_handlers;
 110
 111   /**
 112    * Ports open.
 113    */
 114   const uint32_t *ports;
 115
 116   /**
 117    * Number of ports.
 118    */
 119   unsigned int n_ports;
 120
 121     /**
 122      * Double linked list of the channels this client is connected to, head.
 123      */
 124   struct GNUNET_MESH_Channel *channels_head;
 125
 126     /**
 127      * Double linked list of the channels this client is connected to, tail.
 128      */
 129   struct GNUNET_MESH_Channel *channels_tail;
 130
 131     /**
 132      * Callback for inbound channel creation
 133      */
 134   GNUNET_MESH_InboundChannelNotificationHandler *new_channel;
 135
 136     /**
 137      * Callback for inbound channel disconnection
 138      */
 139   GNUNET_MESH_ChannelEndHandler *cleaner;
 140
 141     /**
 142      * Handle to cancel pending transmissions in case of disconnection
 143      */
 144   struct GNUNET_CLIENT_TransmitHandle *th;
 145
 146     /**
 147      * Closure for all the handlers given by the client
 148      */
 149   void *cls;
 150
 151     /**
 152      * Messages to send to the service, head.
 153      */
 154   struct GNUNET_MESH_TransmitHandle *th_head;
 155
 156     /**
 157      * Messages to send to the service, tail.
 158      */
 159   struct GNUNET_MESH_TransmitHandle *th_tail;
 160
 161     /**
 162      * chid of the next channel to create (to avoid reusing IDs often)
 163      */
 164   MESH_ChannelNumber next_chid;
 165
 166     /**
 167      * Have we started the task to receive messages from the service
 168      * yet? We do this after we send the 'MESH_LOCAL_CONNECT' message.
 169      */
 170   int in_receive;
 171
 172   /**
 173    * Configuration given by the client, in case of reconnection
 174    */
 175   const struct GNUNET_CONFIGURATION_Handle *cfg;
 176
 177   /**
 178    * Time to the next reconnect in case one reconnect fails
 179    */
 180   struct GNUNET_TIME_Relative reconnect_time;
 181
 182   /**
 183    * Task for trying to reconnect.
 184    */
 185   GNUNET_SCHEDULER_TaskIdentifier reconnect_task;
 186
 187   /**
 188    * Monitor callback
 189    */
 190   GNUNET_MESH_ChannelsCB channels_cb;
 191
 192   /**
 193    * Monitor callback closure.
 194    */
 195   void *channels_cls;
 196
 197   /**
 198    * Channel callback.
 199    */
 200   GNUNET_MESH_ChannelCB channel_cb;
 201
 202   /**
 203    * Channel callback closure.
 204    */
 205   void *channel_cls;
 206
 207   /**
 208    * Monitor callback
 209    */
 210   GNUNET_MESH_PeersCB peers_cb;
 211
 212   /**
 213    * Monitor callback closure.
 214    */
 215   void *peers_cls;
 216
 217   /**
 218    * Monitor callback
 219    */
 220   GNUNET_MESH_TunnelsCB tunnels_cb;
 221
 222   /**
 223    * Monitor callback closure.
 224    */
 225   void *tunnels_cls;
 226
 227   /**
 228    * Tunnel callback.
 229    */
 230   GNUNET_MESH_TunnelCB tunnel_cb;
 231
 232   /**
 233    * Tunnel callback closure.
 234    */
 235   void *tunnel_cls;
 236 };
 237
 238
 239 /**
 240  * Description of a peer
 241  */
 242 struct GNUNET_MESH_Peer
 243 {
 244     /**
 245      * ID of the peer in short form
 246      */
 247   GNUNET_PEER_Id id;
 248
 249   /**
 250    * Channel this peer belongs to
 251    */
 252   struct GNUNET_MESH_Channel *t;
 253 };
 254
 255
 256 /**
 257  * Opaque handle to a channel.
 258  */
 259 struct GNUNET_MESH_Channel
 260 {
 261
 262     /**
 263      * DLL next
 264      */
 265   struct GNUNET_MESH_Channel *next;
 266
 267     /**
 268      * DLL prev
 269      */
 270   struct GNUNET_MESH_Channel *prev;
 271
 272     /**
 273      * Handle to the mesh this channel belongs to
 274      */
 275   struct GNUNET_MESH_Handle *mesh;
 276
 277     /**
 278      * Local ID of the channel
 279      */
 280   MESH_ChannelNumber chid;
 281
 282     /**
 283      * Port number.
 284      */
 285   uint32_t port;
 286
 287     /**
 288      * Other end of the channel.
 289      */
 290   GNUNET_PEER_Id peer;
 291
 292   /**
 293    * Any data the caller wants to put in here
 294    */
 295   void *ctx;
 296
 297     /**
 298      * Size of packet queued in this channel
 299      */
 300   unsigned int packet_size;
 301
 302     /**
 303      * Channel options: reliability, etc.
 304      */
 305   enum GNUNET_MESH_ChannelOption options;
 306
 307     /**
 308      * Are we allowed to send to the service?
 309      */
 310   int allow_send;
 311
 312 };
 313
 314
 315 /**
 316  * Implementation state for mesh's message queue.
 317  */
 318 struct MeshMQState
 319 {
 320   /**
 321    * The current transmit handle, or NULL
 322    * if no transmit is active.
 323    */
 324   struct GNUNET_MESH_TransmitHandle *th;
 325
 326   /**
 327    * Channel to send the data over.
 328    */
 329   struct GNUNET_MESH_Channel *channel;
 330 };
 331
 332
 333 /******************************************************************************/
 334 /***********************         DECLARATIONS         *************************/
 335 /******************************************************************************/
 336
 337 /**
 338  * Function called to send a message to the service.
 339  * "buf" will be NULL and "size" zero if the socket was closed for writing in
 340  * the meantime.
 341  *
 342  * @param cls closure, the mesh handle
 343  * @param size number of bytes available in buf
 344  * @param buf where the callee should write the connect message
 345  * @return number of bytes written to buf
 346  */
 347 static size_t
 348 send_callback (void *cls, size_t size, void *buf);
 349
 350
 351 /******************************************************************************/
 352 /***********************     AUXILIARY FUNCTIONS      *************************/
 353 /******************************************************************************/
 354
 355 /**
 356  * Check if transmission is a payload packet.
 357  *
 358  * @param th Transmission handle.
 359  *
 360  * @return GNUNET_YES if it is a payload packet,
 361  *         GNUNET_NO if it is a mesh management packet.
 362  */
 363 static int
 364 th_is_payload (struct GNUNET_MESH_TransmitHandle *th)
 365 {
 366   return (th->notify != NULL) ? GNUNET_YES : GNUNET_NO;
 367 }
 368
 369
 370 /**
 371  * Check whether there is any message ready in the queue and find the size.
 372  *
 373  * @param h Mesh handle.
 374  *
 375  * @return The size of the first ready message in the queue,
 376  *         0 if there is none.
 377  */
 378 static size_t
 379 message_ready_size (struct GNUNET_MESH_Handle *h)
 380 {
 381   struct GNUNET_MESH_TransmitHandle *th;
 382   struct GNUNET_MESH_Channel *ch;
 383
 384   for (th = h->th_head; NULL != th; th = th->next)
 385   {
 386     ch = th->channel;
 387     if (GNUNET_NO == th_is_payload (th))
 388     {
 389       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  message internal\n");
 390       return th->size;
 391     }
 392     if (GNUNET_YES == ch->allow_send)
 393     {
 394       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  message payload ok\n");
 395       return th->size;
 396     }
 397   }
 398   return 0;
 399 }
 400
 401
 402 /**
 403  * Get the channel handler for the channel specified by id from the given handle
 404  * @param h Mesh handle
 405  * @param chid ID of the wanted channel
 406  * @return handle to the required channel or NULL if not found
 407  */
 408 static struct GNUNET_MESH_Channel *
 409 retrieve_channel (struct GNUNET_MESH_Handle *h, MESH_ChannelNumber chid)
 410 {
 411   struct GNUNET_MESH_Channel *ch;
 412
 413   ch = h->channels_head;
 414   while (ch != NULL)
 415   {
 416     if (ch->chid == chid)
 417       return ch;
 418     ch = ch->next;
 419   }
 420   return NULL;
 421 }
 422
 423
 424 /**
 425  * Create a new channel and insert it in the channel list of the mesh handle
 426  *
 427  * @param h Mesh handle
 428  * @param chid Desired chid of the channel, 0 to assign one automatically.
 429  *
 430  * @return Handle to the created channel.
 431  */
 432 static struct GNUNET_MESH_Channel *
 433 create_channel (struct GNUNET_MESH_Handle *h, MESH_ChannelNumber chid)
 434 {
 435   struct GNUNET_MESH_Channel *ch;
 436
 437   ch = GNUNET_new (struct GNUNET_MESH_Channel);
 438   GNUNET_CONTAINER_DLL_insert (h->channels_head, h->channels_tail, ch);
 439   ch->mesh = h;
 440   if (0 == chid)
 441   {
 442     ch->chid = h->next_chid;
 443     while (NULL != retrieve_channel (h, h->next_chid))
 444     {
 445       h->next_chid++;
 446       h->next_chid &= ~GNUNET_MESH_LOCAL_CHANNEL_ID_SERV;
 447       h->next_chid |= GNUNET_MESH_LOCAL_CHANNEL_ID_CLI;
 448     }
 449   }
 450   else
 451   {
 452     ch->chid = chid;
 453   }
 454   ch->allow_send = GNUNET_NO;
 455   return ch;
 456 }
 457
 458
 459 /**
 460  * Destroy the specified channel.
 461  * - Destroys all peers, calling the disconnect callback on each if needed
 462  * - Cancels all outgoing traffic for that channel, calling respective notifys
 463  * - Calls cleaner if channel was inbound
 464  * - Frees all memory used
 465  *
 466  * @param ch Pointer to the channel.
 467  * @param call_cleaner Whether to call the cleaner handler.
 468  *
 469  * @return Handle to the required channel or NULL if not found.
 470  */
 471 static void
 472 destroy_channel (struct GNUNET_MESH_Channel *ch, int call_cleaner)
 473 {
 474   struct GNUNET_MESH_Handle *h;
 475   struct GNUNET_MESH_TransmitHandle *th;
 476   struct GNUNET_MESH_TransmitHandle *next;
 477
 478   LOG (GNUNET_ERROR_TYPE_DEBUG, " destroy_channel %X\n", ch->chid);
 479
 480   if (NULL == ch)
 481   {
 482     GNUNET_break (0);
 483     return;
 484   }
 485   h = ch->mesh;
 486
 487   GNUNET_CONTAINER_DLL_remove (h->channels_head, h->channels_tail, ch);
 488
 489   /* signal channel destruction */
 490   if ( (NULL != h->cleaner) && (0 != ch->peer) && (GNUNET_YES == call_cleaner) )
 491   {
 492     LOG (GNUNET_ERROR_TYPE_DEBUG, " calling cleaner\n");
 493     h->cleaner (h->cls, ch, ch->ctx);
 494   }
 495
 496   /* check that clients did not leave messages behind in the queue */
 497   for (th = h->th_head; NULL != th; th = next)
 498   {
 499     next = th->next;
 500     if (th->channel != ch)
 501       continue;
 502     /* Clients should have aborted their requests already.
 503      * Management traffic should be ok, as clients can't cancel that.
 504      * If the service crashed and we are reconnecting, it's ok.
 505      */
 506     GNUNET_break (GNUNET_NO == th_is_payload (th)
 507                   || GNUNET_NO == h->in_receive);
 508     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
 509
 510     /* clean up request */
 511     if (GNUNET_SCHEDULER_NO_TASK != th->timeout_task)
 512       GNUNET_SCHEDULER_cancel (th->timeout_task);
 513     GNUNET_free (th);
 514   }
 515
 516   /* if there are no more pending requests with mesh service, cancel active request */
 517   /* Note: this should be unnecessary... */
 518   if ((0 == message_ready_size (h)) && (NULL != h->th))
 519   {
 520     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 521     h->th = NULL;
 522   }
 523
 524   if (0 != ch->peer)
 525     GNUNET_PEER_change_rc (ch->peer, -1);
 526   GNUNET_free (ch);
 527   return;
 528 }
 529
 530
 531 /**
 532  * Notify client that the transmission has timed out
 533  *
 534  * @param cls closure
 535  * @param tc task context
 536  */
 537 static void
 538 timeout_transmission (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 539 {
 540   struct GNUNET_MESH_TransmitHandle *th = cls;
 541   struct GNUNET_MESH_Handle *mesh;
 542
 543   mesh = th->channel->mesh;
 544   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
 545   th->channel->packet_size = 0;
 546   if (GNUNET_YES == th_is_payload (th))
 547     th->notify (th->notify_cls, 0, NULL);
 548   GNUNET_free (th);
 549   if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
 550   {
 551     /* nothing ready to transmit, no point in asking for transmission */
 552     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
 553     mesh->th = NULL;
 554   }
 555 }
 556
 557
 558 /**
 559  * Add a transmit handle to the transmission queue and set the
 560  * timeout if needed.
 561  *
 562  * @param h mesh handle with the queue head and tail
 563  * @param th handle to the packet to be transmitted
 564  */
 565 static void
 566 add_to_queue (struct GNUNET_MESH_Handle *h,
 567               struct GNUNET_MESH_TransmitHandle *th)
 568 {
 569   GNUNET_CONTAINER_DLL_insert_tail (h->th_head, h->th_tail, th);
 570   if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us == th->timeout.abs_value_us)
 571     return;
 572   th->timeout_task =
 573       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
 574                                     (th->timeout), &timeout_transmission, th);
 575 }
 576
 577
 578 /**
 579  * Auxiliary function to send an already constructed packet to the service.
 580  * Takes care of creating a new queue element, copying the message and
 581  * calling the tmt_rdy function if necessary.
 582  *
 583  * @param h mesh handle
 584  * @param msg message to transmit
 585  * @param channel channel this send is related to (NULL if N/A)
 586  */
 587 static void
 588 send_packet (struct GNUNET_MESH_Handle *h,
 589              const struct GNUNET_MessageHeader *msg,
 590              struct GNUNET_MESH_Channel *channel);
 591
 592
 593 /**
 594  * Send an ack on the channel to confirm the processing of a message.
 595  *
 596  * @param ch Channel on which to send the ACK.
 597  */
 598 static void
 599 send_ack (struct GNUNET_MESH_Channel *ch)
 600 {
 601   struct GNUNET_MESH_LocalAck msg;
 602
 603   LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending ACK on channel %X\n", ch->chid);
 604   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
 605   msg.header.size = htons (sizeof (msg));
 606   msg.channel_id = htonl (ch->chid);
 607
 608   send_packet (ch->mesh, &msg.header, ch);
 609   return;
 610 }
 611
 612
 613
 614 /**
 615  * Reconnect callback: tries to reconnect again after a failer previous
 616  * reconnecttion
 617  * @param cls closure (mesh handle)
 618  * @param tc task context
 619  */
 620 static void
 621 reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 622
 623
 624 /**
 625  * Send a connect packet to the service with the applications and types
 626  * requested by the user.
 627  *
 628  * @param h The mesh handle.
 629  *
 630  */
 631 static void
 632 send_connect (struct GNUNET_MESH_Handle *h)
 633 {
 634   size_t size;
 635
 636   size = sizeof (struct GNUNET_MESH_ClientConnect);
 637   size += h->n_ports * sizeof (uint32_t);
 638   {
 639     char buf[size] GNUNET_ALIGN;
 640     struct GNUNET_MESH_ClientConnect *msg;
 641     uint32_t *ports;
 642     uint16_t i;
 643
 644     /* build connection packet */
 645     msg = (struct GNUNET_MESH_ClientConnect *) buf;
 646     msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT);
 647     msg->header.size = htons (size);
 648     ports = (uint32_t *) &msg[1];
 649     for (i = 0; i < h->n_ports; i++)
 650     {
 651       ports[i] = htonl (h->ports[i]);
 652       LOG (GNUNET_ERROR_TYPE_DEBUG, " port %u\n",
 653            h->ports[i]);
 654     }
 655     LOG (GNUNET_ERROR_TYPE_DEBUG,
 656          "Sending %lu bytes long message with %u ports\n",
 657          ntohs (msg->header.size), h->n_ports);
 658     send_packet (h, &msg->header, NULL);
 659   }
 660 }
 661
 662
 663 /**
 664  * Reconnect to the service, retransmit all infomation to try to restore the
 665  * original state.
 666  *
 667  * @param h handle to the mesh
 668  *
 669  * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
 670  */
 671 static int
 672 do_reconnect (struct GNUNET_MESH_Handle *h)
 673 {
 674   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 675   LOG (GNUNET_ERROR_TYPE_DEBUG, "*******   RECONNECT   *******\n");
 676   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 677   LOG (GNUNET_ERROR_TYPE_DEBUG, "******** on %p *******\n", h);
 678   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 679
 680   /* disconnect */
 681   if (NULL != h->th)
 682   {
 683     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 684     h->th = NULL;
 685   }
 686   if (NULL != h->client)
 687   {
 688     GNUNET_CLIENT_disconnect (h->client);
 689   }
 690
 691   /* connect again */
 692   h->client = GNUNET_CLIENT_connect ("mesh", h->cfg);
 693   if (h->client == NULL)
 694   {
 695     h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 696                                                       &reconnect_cbk, h);
 697     h->reconnect_time =
 698         GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
 699                                   GNUNET_TIME_relative_multiply
 700                                   (h->reconnect_time, 2));
 701     LOG (GNUNET_ERROR_TYPE_DEBUG, "Next retry in %s\n",
 702          GNUNET_STRINGS_relative_time_to_string (h->reconnect_time,
 703                                                  GNUNET_NO));
 704     GNUNET_break (0);
 705     return GNUNET_NO;
 706   }
 707   else
 708   {
 709     h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
 710   }
 711   send_connect (h);
 712   return GNUNET_YES;
 713 }
 714
 715 /**
 716  * Reconnect callback: tries to reconnect again after a failer previous
 717  * reconnecttion
 718  * @param cls closure (mesh handle)
 719  * @param tc task context
 720  */
 721 static void
 722 reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 723 {
 724   struct GNUNET_MESH_Handle *h = cls;
 725
 726   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 727   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 728     return;
 729   do_reconnect (h);
 730 }
 731
 732
 733 /**
 734  * Reconnect to the service, retransmit all infomation to try to restore the
 735  * original state.
 736  *
 737  * @param h handle to the mesh
 738  *
 739  * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
 740  */
 741 static void
 742 reconnect (struct GNUNET_MESH_Handle *h)
 743 {
 744   struct GNUNET_MESH_Channel *ch;
 745   struct GNUNET_MESH_Channel *next;
 746
 747   LOG (GNUNET_ERROR_TYPE_DEBUG,
 748        "Requested RECONNECT, destroying all channels\n");
 749   h->in_receive = GNUNET_NO;
 750   for (ch = h->channels_head; NULL != ch; ch = next)
 751   {
 752     next = ch->next;
 753     destroy_channel (ch, GNUNET_YES);
 754   }
 755   if (GNUNET_SCHEDULER_NO_TASK == h->reconnect_task)
 756     h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 757                                                       &reconnect_cbk, h);
 758 }
 759
 760
 761 /******************************************************************************/
 762 /***********************      RECEIVE HANDLERS     ****************************/
 763 /******************************************************************************/
 764
 765 /**
 766  * Process the new channel notification and add it to the channels in the handle
 767  *
 768  * @param h     The mesh handle
 769  * @param msg   A message with the details of the new incoming channel
 770  */
 771 static void
 772 process_channel_created (struct GNUNET_MESH_Handle *h,
 773                         const struct GNUNET_MESH_ChannelMessage *msg)
 774 {
 775   struct GNUNET_MESH_Channel *ch;
 776   MESH_ChannelNumber chid;
 777   uint32_t port;
 778
 779   chid = ntohl (msg->channel_id);
 780   port = ntohl (msg->port);
 781   LOG (GNUNET_ERROR_TYPE_DEBUG, "Creating incoming channel %X:%u\n", chid, port);
 782   if (chid < GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
 783   {
 784     GNUNET_break (0);
 785     return;
 786   }
 787   if (NULL != h->new_channel)
 788   {
 789     void *ctx;
 790
 791     ch = create_channel (h, chid);
 792     ch->allow_send = GNUNET_NO;
 793     ch->peer = GNUNET_PEER_intern (&msg->peer);
 794     ch->mesh = h;
 795     ch->chid = chid;
 796     ch->port = port;
 797     ch->options = ntohl (msg->opt);
 798
 799     LOG (GNUNET_ERROR_TYPE_DEBUG, "  created channel %p\n", ch);
 800     ctx = h->new_channel (h->cls, ch, &msg->peer, ch->port, ch->options);
 801     if (NULL != ctx)
 802       ch->ctx = ctx;
 803     LOG (GNUNET_ERROR_TYPE_DEBUG, "User notified\n");
 804   }
 805   else
 806   {
 807     struct GNUNET_MESH_ChannelMessage d_msg;
 808
 809     LOG (GNUNET_ERROR_TYPE_DEBUG, "No handler for incoming channels\n");
 810
 811     d_msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY);
 812     d_msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
 813     d_msg.channel_id = msg->channel_id;
 814     memset (&d_msg.peer, 0, sizeof (struct GNUNET_PeerIdentity));
 815     d_msg.port = 0;
 816     d_msg.opt = 0;
 817
 818     send_packet (h, &d_msg.header, NULL);
 819   }
 820   return;
 821 }
 822
 823
 824 /**
 825  * Process the channel destroy notification and free associated resources
 826  *
 827  * @param h     The mesh handle
 828  * @param msg   A message with the details of the channel being destroyed
 829  */
 830 static void
 831 process_channel_destroy (struct GNUNET_MESH_Handle *h,
 832                          const struct GNUNET_MESH_ChannelMessage *msg)
 833 {
 834   struct GNUNET_MESH_Channel *ch;
 835   MESH_ChannelNumber chid;
 836
 837   LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel Destroy received from service\n");
 838   chid = ntohl (msg->channel_id);
 839   ch = retrieve_channel (h, chid);
 840
 841   if (NULL == ch)
 842   {
 843     LOG (GNUNET_ERROR_TYPE_DEBUG, "channel %X unknown\n", chid);
 844     return;
 845   }
 846   LOG (GNUNET_ERROR_TYPE_DEBUG, " destroying channel %X\n", ch->chid);
 847   destroy_channel (ch, GNUNET_YES);
 848 }
 849
 850
 851 /**
 852  * Process the incoming data packets, call appropriate handlers.
 853  *
 854  * @param h         The mesh handle
 855  * @param message   A message encapsulating the data
 856  */
 857 static void
 858 process_incoming_data (struct GNUNET_MESH_Handle *h,
 859                        const struct GNUNET_MessageHeader *message)
 860 {
 861   const struct GNUNET_MessageHeader *payload;
 862   const struct GNUNET_MESH_MessageHandler *handler;
 863   struct GNUNET_MESH_LocalData *dmsg;
 864   struct GNUNET_MESH_Channel *ch;
 865   size_t size;
 866   unsigned int i;
 867   uint16_t type;
 868
 869   LOG (GNUNET_ERROR_TYPE_DEBUG, "Got a data message!\n");
 870   dmsg = (struct GNUNET_MESH_LocalData *) message;
 871   ch = retrieve_channel (h, ntohl (dmsg->id));
 872   payload = (struct GNUNET_MessageHeader *) &dmsg[1];
 873   LOG (GNUNET_ERROR_TYPE_DEBUG, "  %s data on channel %s [%X]\n",
 874        GM_f2s (ch->chid >= GNUNET_MESH_LOCAL_CHANNEL_ID_SERV),
 875        GNUNET_i2s (GNUNET_PEER_resolve2 (ch->peer)), ntohl (dmsg->id));
 876
 877   size = ntohs (message->size);
 878   LOG (GNUNET_ERROR_TYPE_DEBUG, "  %u bytes\n", size);
 879
 880   if (NULL == ch)
 881   {
 882     /* Channel was ignored/destroyed, probably service didn't get it yet */
 883     LOG (GNUNET_ERROR_TYPE_DEBUG, "  ignored!\n");
 884     return;
 885   }
 886   type = ntohs (payload->type);
 887   size = ntohs (payload->size);
 888   LOG (GNUNET_ERROR_TYPE_DEBUG, "  payload type %s\n", GM_m2s (type));
 889   for (i = 0; i < h->n_handlers; i++)
 890   {
 891     handler = &h->message_handlers[i];
 892     LOG (GNUNET_ERROR_TYPE_DEBUG, "    checking handler for type %u\n",
 893          handler->type);
 894     if (handler->type == type)
 895     {
 896       if (GNUNET_OK !=
 897           handler->callback (h->cls, ch, &ch->ctx, payload))
 898       {
 899         LOG (GNUNET_ERROR_TYPE_DEBUG, "callback caused disconnection\n");
 900         GNUNET_MESH_channel_destroy (ch);
 901         return;
 902       }
 903       else
 904       {
 905         LOG (GNUNET_ERROR_TYPE_DEBUG,
 906              "callback completed successfully\n");
 907         return;
 908       }
 909     }
 910   }
 911 }
 912
 913
 914 /**
 915  * Process a local ACK message, enabling the client to send
 916  * more data to the service.
 917  *
 918  * @param h Mesh handle.
 919  * @param message Message itself.
 920  */
 921 static void
 922 process_ack (struct GNUNET_MESH_Handle *h,
 923              const struct GNUNET_MessageHeader *message)
 924 {
 925   struct GNUNET_MESH_LocalAck *msg;
 926   struct GNUNET_MESH_Channel *ch;
 927   MESH_ChannelNumber chid;
 928
 929   LOG (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK!\n");
 930   msg = (struct GNUNET_MESH_LocalAck *) message;
 931   chid = ntohl (msg->channel_id);
 932   ch = retrieve_channel (h, chid);
 933   if (NULL == ch)
 934   {
 935     LOG (GNUNET_ERROR_TYPE_DEBUG, "ACK on unknown channel %X\n", chid);
 936     return;
 937   }
 938   LOG (GNUNET_ERROR_TYPE_DEBUG, "  on channel %X!\n", ch->chid);
 939   ch->allow_send = GNUNET_YES;
 940   if (NULL == h->th && 0 < ch->packet_size)
 941   {
 942     LOG (GNUNET_ERROR_TYPE_DEBUG, "  tmt rdy was NULL, requesting!\n");
 943     h->th = GNUNET_CLIENT_notify_transmit_ready (h->client, ch->packet_size,
 944                                                  GNUNET_TIME_UNIT_FOREVER_REL,
 945                                                  GNUNET_YES, &send_callback, h);
 946   }
 947 }
 948
 949
 950 /*
 951  * Process a local reply about info on all channels, pass info to the user.
 952  *
 953  * @param h Mesh handle.
 954  * @param message Message itself.
 955  */
 956 // static void
 957 // process_get_channels (struct GNUNET_MESH_Handle *h,
 958 //                      const struct GNUNET_MessageHeader *message)
 959 // {
 960 //   struct GNUNET_MESH_LocalInfo *msg;
 961 //
 962 //   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Get Channels messasge received\n");
 963 //
 964 //   if (NULL == h->channels_cb)
 965 //   {
 966 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "  ignored\n");
 967 //     return;
 968 //   }
 969 //
 970 //   msg = (struct GNUNET_MESH_LocalInfo *) message;
 971 //   if (ntohs (message->size) !=
 972 //       (sizeof (struct GNUNET_MESH_LocalInfo) +
 973 //        sizeof (struct GNUNET_PeerIdentity)))
 974 //   {
 975 //     GNUNET_break_op (0);
 976 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 977 //                 "Get channels message: size %hu - expected %u\n",
 978 //                 ntohs (message->size),
 979 //                 sizeof (struct GNUNET_MESH_LocalInfo));
 980 //     return;
 981 //   }
 982 //   h->channels_cb (h->channels_cls,
 983 //                   ntohl (msg->channel_id),
 984 //                   &msg->owner,
 985 //                   &msg->destination);
 986 // }
 987
 988
 989
 990 /*
 991  * Process a local monitor_channel reply, pass info to the user.
 992  *
 993  * @param h Mesh handle.
 994  * @param message Message itself.
 995  */
 996 // static void
 997 // process_show_channel (struct GNUNET_MESH_Handle *h,
 998 //                      const struct GNUNET_MessageHeader *message)
 999 // {
1000 //   struct GNUNET_MESH_LocalInfo *msg;
1001 //   size_t esize;
1002 //
1003 //   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Show Channel messasge received\n");
1004 //
1005 //   if (NULL == h->channel_cb)
1006 //   {
1007 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "  ignored\n");
1008 //     return;
1009 //   }
1010 //
1011 //   /* Verify message sanity */
1012 //   msg = (struct GNUNET_MESH_LocalInfo *) message;
1013 //   esize = sizeof (struct GNUNET_MESH_LocalInfo);
1014 //   if (ntohs (message->size) != esize)
1015 //   {
1016 //     GNUNET_break_op (0);
1017 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
1018 //                 "Show channel message: size %hu - expected %u\n",
1019 //                 ntohs (message->size),
1020 //                 esize);
1021 //
1022 //     h->channel_cb (h->channel_cls, NULL, NULL);
1023 //     h->channel_cb = NULL;
1024 //     h->channel_cls = NULL;
1025 //
1026 //     return;
1027 //   }
1028 //
1029 //   h->channel_cb (h->channel_cls,
1030 //                  &msg->destination,
1031 //                  &msg->owner);
1032 // }
1033
1034
1035
1036 /**
1037  * Process a local reply about info on all tunnels, pass info to the user.
1038  *
1039  * @param h Mesh handle.
1040  * @param message Message itself.
1041  */
1042 static void
1043 process_get_peers (struct GNUNET_MESH_Handle *h,
1044                      const struct GNUNET_MessageHeader *message)
1045 {
1046   struct GNUNET_MESH_LocalInfoPeer *msg;
1047   uint16_t size;
1048
1049   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Get Peer messasge received\n");
1050
1051   if (NULL == h->peers_cb)
1052   {
1053     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ignored\n");
1054     return;
1055   }
1056
1057   size = ntohs (message->size);
1058   if (sizeof (struct GNUNET_MESH_LocalInfoPeer) > size)
1059   {
1060     h->peers_cb (h->peers_cls, NULL, -1, 0, 0);
1061     h->peers_cb = NULL;
1062     h->peers_cls = NULL;
1063     return;
1064   }
1065
1066   msg = (struct GNUNET_MESH_LocalInfoPeer *) message;
1067   h->peers_cb (h->peers_cls, &msg->destination,
1068                (int) ntohs (msg->tunnel), (unsigned int ) ntohs (msg->paths),
1069                0);
1070 }
1071
1072
1073 /**
1074  * Process a local reply about info on all tunnels, pass info to the user.
1075  *
1076  * @param h Mesh handle.
1077  * @param message Message itself.
1078  */
1079 static void
1080 process_get_tunnels (struct GNUNET_MESH_Handle *h,
1081                      const struct GNUNET_MessageHeader *message)
1082 {
1083   struct GNUNET_MESH_LocalInfoTunnel *msg;
1084   uint16_t size;
1085
1086   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Get Tunnels messasge received\n");
1087
1088   if (NULL == h->tunnels_cb)
1089   {
1090     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ignored\n");
1091     return;
1092   }
1093
1094   size = ntohs (message->size);
1095   if (sizeof (struct GNUNET_MESH_LocalInfoTunnel) > size)
1096   {
1097     h->tunnels_cb (h->tunnel_cls, NULL, 0, 0, 0, 0);
1098     h->tunnels_cb = NULL;
1099     h->tunnels_cls = NULL;
1100     return;
1101   }
1102
1103   msg = (struct GNUNET_MESH_LocalInfoTunnel *) message;
1104   h->tunnels_cb (h->tunnel_cls,
1105                  &msg->destination,
1106                  ntohl (msg->channels),
1107                  ntohl (msg->connections),
1108                  ntohs (msg->estate),
1109                  ntohs (msg->cstate));
1110
1111 }
1112
1113
1114
1115 /**
1116  * Process a local tunnel info reply, pass info to the user.
1117  *
1118  * @param h Mesh handle.
1119  * @param message Message itself.
1120  */
1121 static void
1122 process_get_tunnel (struct GNUNET_MESH_Handle *h,
1123                     const struct GNUNET_MessageHeader *message)
1124 {
1125   struct GNUNET_MESH_LocalInfoTunnel *msg;
1126   size_t esize;
1127   size_t msize;
1128   unsigned int ch_n;
1129   unsigned int c_n;
1130   struct GNUNET_HashCode *conns;
1131   MESH_ChannelNumber *chns;
1132
1133   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Get Tunnel messasge received\n");
1134   if (NULL == h->tunnel_cb)
1135   {
1136     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "  ignored\n");
1137     return;
1138   }
1139
1140   /* Verify message sanity */
1141   msg = (struct GNUNET_MESH_LocalInfoTunnel *) message;
1142   msize = ntohs (message->size);
1143   esize = sizeof (struct GNUNET_MESH_LocalInfoTunnel);
1144   if (esize > msize)
1145   {
1146     GNUNET_break_op (0);
1147     h->tunnel_cb (h->tunnel_cls, NULL, 0, 0, NULL, NULL, 0, 0);
1148     goto clean_cls;
1149   }
1150   ch_n = ntohl (msg->channels);
1151   c_n = ntohl (msg->connections);
1152   esize += ch_n * sizeof (MESH_ChannelNumber);
1153   esize += c_n * sizeof (struct GNUNET_HashCode);
1154   if (msize != esize)
1155   {
1156     GNUNET_break_op (0);
1157     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "m:%u, e: %u (%u ch, %u conn)\n",
1158                 msize, esize, ch_n, c_n);
1159     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "%u (%u ch, %u conn)\n",
1160                 sizeof (struct GNUNET_MESH_LocalInfoTunnel),
1161                 sizeof (MESH_ChannelNumber), sizeof (struct GNUNET_HashCode));
1162     h->tunnel_cb (h->tunnel_cls, NULL, 0, 0, NULL, NULL, 0, 0);
1163     goto clean_cls;
1164   }
1165
1166   /* Call Callback with tunnel info. */
1167   conns = (struct GNUNET_HashCode *) &msg[1];
1168   chns = (MESH_ChannelNumber *) &conns[c_n];
1169   h->tunnel_cb (h->tunnel_cls, &msg->destination,
1170                 ch_n, c_n, chns, conns,
1171                 ntohs (msg->estate), ntohs (msg->cstate));
1172
1173 clean_cls:
1174   h->tunnel_cb = NULL;
1175   h->tunnel_cls = NULL;
1176 }
1177
1178 /**
1179  * Function to process all messages received from the service
1180  *
1181  * @param cls closure
1182  * @param msg message received, NULL on timeout or fatal error
1183  */
1184 static void
1185 msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
1186 {
1187   struct GNUNET_MESH_Handle *h = cls;
1188   uint16_t type;
1189
1190   if (msg == NULL)
1191   {
1192     LOG (GNUNET_ERROR_TYPE_DEBUG,
1193          "Mesh service disconnected, reconnecting\n", h);
1194     reconnect (h);
1195     return;
1196   }
1197   type = ntohs (msg->type);
1198   LOG (GNUNET_ERROR_TYPE_DEBUG, "\n");
1199   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a message: %s\n",
1200        GM_m2s (type));
1201   switch (type)
1202   {
1203     /* Notify of a new incoming channel */
1204   case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
1205     process_channel_created (h, (struct GNUNET_MESH_ChannelMessage *) msg);
1206     break;
1207     /* Notify of a channel disconnection */
1208   case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY: /* TODO separate(gid problem)*/
1209   case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_NACK:
1210     process_channel_destroy (h, (struct GNUNET_MESH_ChannelMessage *) msg);
1211     break;
1212   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_DATA:
1213     process_incoming_data (h, msg);
1214     break;
1215   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
1216     process_ack (h, msg);
1217     break;
1218 //   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS:
1219 //     process_get_channels (h, msg);
1220 //     break;
1221 //   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL:
1222 //     process_show_channel (h, msg);
1223 //     break;
1224   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_PEERS:
1225     process_get_peers (h, msg);
1226     break;
1227   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNELS:
1228     process_get_tunnels (h, msg);
1229     break;
1230   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNEL:
1231     process_get_tunnel (h, msg);
1232     break;
1233 //   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL:
1234 //     process_show_channel (h, msg);
1235 //     break;
1236   default:
1237     /* We shouldn't get any other packages, log and ignore */
1238     LOG (GNUNET_ERROR_TYPE_WARNING,
1239          "unsolicited message form service (type %s)\n",
1240          GM_m2s (ntohs (msg->type)));
1241   }
1242   LOG (GNUNET_ERROR_TYPE_DEBUG, "message processed\n");
1243   if (GNUNET_YES == h->in_receive)
1244   {
1245     GNUNET_CLIENT_receive (h->client, &msg_received, h,
1246                            GNUNET_TIME_UNIT_FOREVER_REL);
1247   }
1248   else
1249   {
1250     LOG (GNUNET_ERROR_TYPE_DEBUG,
1251          "in receive off, not calling CLIENT_receive\n");
1252   }
1253 }
1254
1255
1256 /******************************************************************************/
1257 /************************       SEND FUNCTIONS     ****************************/
1258 /******************************************************************************/
1259
1260 /**
1261  * Function called to send a message to the service.
1262  * "buf" will be NULL and "size" zero if the socket was closed for writing in
1263  * the meantime.
1264  *
1265  * @param cls closure, the mesh handle
1266  * @param size number of bytes available in buf
1267  * @param buf where the callee should write the connect message
1268  * @return number of bytes written to buf
1269  */
1270 static size_t
1271 send_callback (void *cls, size_t size, void *buf)
1272 {
1273   struct GNUNET_MESH_Handle *h = cls;
1274   struct GNUNET_MESH_TransmitHandle *th;
1275   struct GNUNET_MESH_TransmitHandle *next;
1276   struct GNUNET_MESH_Channel *ch;
1277   char *cbuf = buf;
1278   size_t tsize;
1279   size_t psize;
1280   size_t nsize;
1281
1282   LOG (GNUNET_ERROR_TYPE_DEBUG, "\n");
1283   LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() Buffer %u\n", size);
1284   if ((0 == size) || (NULL == buf))
1285   {
1286     LOG (GNUNET_ERROR_TYPE_DEBUG, "# Received NULL send callback on %p\n", h);
1287     reconnect (h);
1288     h->th = NULL;
1289     return 0;
1290   }
1291   tsize = 0;
1292   next = h->th_head;
1293   nsize = message_ready_size (h);
1294   while ((NULL != (th = next)) && (0 < nsize) && (size >= nsize))
1295   {
1296     ch = th->channel;
1297     if (GNUNET_YES == th_is_payload (th))
1298     {
1299       struct GNUNET_MESH_LocalData *dmsg;
1300       struct GNUNET_MessageHeader *mh;
1301
1302       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  payload\n");
1303       if (GNUNET_NO == ch->allow_send)
1304       {
1305         /* This channel is not ready to transmit yet, try next message */
1306         next = th->next;
1307         continue;
1308       }
1309       ch->packet_size = 0;
1310       GNUNET_assert (size >= th->size);
1311       dmsg = (struct GNUNET_MESH_LocalData *) cbuf;
1312       mh = (struct GNUNET_MessageHeader *) &dmsg[1];
1313       psize = th->notify (th->notify_cls,
1314                           size - sizeof (struct GNUNET_MESH_LocalData),
1315                           mh);
1316       if (psize > 0)
1317       {
1318         psize += sizeof (struct GNUNET_MESH_LocalData);
1319         GNUNET_assert (size >= psize);
1320         dmsg->header.size = htons (psize);
1321         dmsg->id = htonl (ch->chid);
1322         dmsg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_DATA);
1323         LOG (GNUNET_ERROR_TYPE_DEBUG, "#  payload type %s\n",
1324              GM_m2s (ntohs (mh->type)));
1325                 ch->allow_send = GNUNET_NO;
1326       }
1327       else
1328       {
1329         LOG (GNUNET_ERROR_TYPE_DEBUG,
1330              "#  callback returned size 0, "
1331              "application canceled transmission\n");
1332       }
1333     }
1334     else
1335     {
1336       struct GNUNET_MessageHeader *mh = (struct GNUNET_MessageHeader *) &th[1];
1337
1338       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  mesh internal traffic, type %s\n",
1339            GM_m2s (ntohs (mh->type)));
1340       memcpy (cbuf, &th[1], th->size);
1341       psize = th->size;
1342     }
1343     if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1344       GNUNET_SCHEDULER_cancel (th->timeout_task);
1345     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
1346     GNUNET_free (th);
1347     next = h->th_head;
1348     nsize = message_ready_size (h);
1349     cbuf += psize;
1350     size -= psize;
1351     tsize += psize;
1352   }
1353   LOG (GNUNET_ERROR_TYPE_DEBUG, "#  total size: %u\n", tsize);
1354   h->th = NULL;
1355   size = message_ready_size (h);
1356   if (0 != size)
1357   {
1358     LOG (GNUNET_ERROR_TYPE_DEBUG, "#  next size: %u\n", size);
1359     h->th =
1360         GNUNET_CLIENT_notify_transmit_ready (h->client, size,
1361                                              GNUNET_TIME_UNIT_FOREVER_REL,
1362                                              GNUNET_YES, &send_callback, h);
1363   }
1364   else
1365   {
1366     if (NULL != h->th_head)
1367       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  can't transmit any more\n");
1368     else
1369       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  nothing left to transmit\n");
1370   }
1371   if (GNUNET_NO == h->in_receive)
1372   {
1373     LOG (GNUNET_ERROR_TYPE_DEBUG, "# start receiving from service\n");
1374     h->in_receive = GNUNET_YES;
1375     GNUNET_CLIENT_receive (h->client, &msg_received, h,
1376                            GNUNET_TIME_UNIT_FOREVER_REL);
1377   }
1378   LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() END\n");
1379   return tsize;
1380 }
1381
1382
1383 /**
1384  * Auxiliary function to send an already constructed packet to the service.
1385  * Takes care of creating a new queue element, copying the message and
1386  * calling the tmt_rdy function if necessary.
1387  *
1388  * @param h mesh handle
1389  * @param msg message to transmit
1390  * @param channel channel this send is related to (NULL if N/A)
1391  */
1392 static void
1393 send_packet (struct GNUNET_MESH_Handle *h,
1394              const struct GNUNET_MessageHeader *msg,
1395              struct GNUNET_MESH_Channel *channel)
1396 {
1397   struct GNUNET_MESH_TransmitHandle *th;
1398   size_t msize;
1399
1400   LOG (GNUNET_ERROR_TYPE_DEBUG, " Sending message to service: %s\n",
1401        GM_m2s(ntohs(msg->type)));
1402   msize = ntohs (msg->size);
1403   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle) + msize);
1404   th->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
1405   th->size = msize;
1406   th->channel = channel;
1407   memcpy (&th[1], msg, msize);
1408   add_to_queue (h, th);
1409   LOG (GNUNET_ERROR_TYPE_DEBUG, "  queued\n");
1410   if (NULL != h->th)
1411     return;
1412   LOG (GNUNET_ERROR_TYPE_DEBUG, "  calling ntfy tmt rdy for %u bytes\n", msize);
1413   h->th =
1414       GNUNET_CLIENT_notify_transmit_ready (h->client, msize,
1415                                            GNUNET_TIME_UNIT_FOREVER_REL,
1416                                            GNUNET_YES, &send_callback, h);
1417 }
1418
1419
1420 /******************************************************************************/
1421 /**********************      API CALL DEFINITIONS     *************************/
1422 /******************************************************************************/
1423
1424 struct GNUNET_MESH_Handle *
1425 GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
1426                      GNUNET_MESH_InboundChannelNotificationHandler new_channel,
1427                      GNUNET_MESH_ChannelEndHandler cleaner,
1428                      const struct GNUNET_MESH_MessageHandler *handlers,
1429                      const uint32_t *ports)
1430 {
1431   struct GNUNET_MESH_Handle *h;
1432
1433   LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect()\n");
1434   h = GNUNET_new (struct GNUNET_MESH_Handle);
1435   LOG (GNUNET_ERROR_TYPE_DEBUG, " addr %p\n", h);
1436   h->cfg = cfg;
1437   h->new_channel = new_channel;
1438   h->cleaner = cleaner;
1439   h->client = GNUNET_CLIENT_connect ("mesh", cfg);
1440   if (h->client == NULL)
1441   {
1442     GNUNET_break (0);
1443     GNUNET_free (h);
1444     return NULL;
1445   }
1446   h->cls = cls;
1447   h->message_handlers = handlers;
1448   h->ports = ports;
1449   h->next_chid = GNUNET_MESH_LOCAL_CHANNEL_ID_CLI;
1450   h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
1451   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1452
1453   if (NULL != ports && ports[0] != 0 && NULL == new_channel)
1454   {
1455     GNUNET_break (0);
1456     LOG (GNUNET_ERROR_TYPE_DEBUG,
1457          "no new channel handler given, ports parameter is useless!!\n");
1458   }
1459   if ((NULL == ports || ports[0] == 0) && NULL != new_channel)
1460   {
1461     GNUNET_break (0);
1462     LOG (GNUNET_ERROR_TYPE_DEBUG,
1463          "no ports given, new channel handler will never be called!!\n");
1464   }
1465   /* count handlers */
1466   for (h->n_handlers = 0;
1467        handlers && handlers[h->n_handlers].type;
1468        h->n_handlers++) ;
1469   for (h->n_ports = 0;
1470        ports && ports[h->n_ports];
1471        h->n_ports++) ;
1472   send_connect (h);
1473   LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect() END\n");
1474   return h;
1475 }
1476
1477
1478 void
1479 GNUNET_MESH_disconnect (struct GNUNET_MESH_Handle *handle)
1480 {
1481   struct GNUNET_MESH_Channel *ch;
1482   struct GNUNET_MESH_Channel *aux;
1483   struct GNUNET_MESH_TransmitHandle *th;
1484
1485   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH DISCONNECT\n");
1486
1487   ch = handle->channels_head;
1488   while (NULL != ch)
1489   {
1490     aux = ch->next;
1491     if (ch->chid < GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
1492     {
1493       GNUNET_break (0);
1494       LOG (GNUNET_ERROR_TYPE_DEBUG, "channel %X not destroyed\n", ch->chid);
1495     }
1496     destroy_channel (ch, GNUNET_YES);
1497     ch = aux;
1498   }
1499   while ( (th = handle->th_head) != NULL)
1500   {
1501     struct GNUNET_MessageHeader *msg;
1502
1503     /* Make sure it is an allowed packet (everything else should have been
1504      * already canceled).
1505      */
1506     GNUNET_break (GNUNET_NO == th_is_payload (th));
1507     msg = (struct GNUNET_MessageHeader *) &th[1];
1508     switch (ntohs(msg->type))
1509     {
1510       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT:
1511       case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
1512       case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
1513       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS:
1514       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL:
1515       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_PEER:
1516       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_PEERS:
1517       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNEL:
1518       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNELS:
1519         break;
1520       default:
1521         GNUNET_break (0);
1522         LOG (GNUNET_ERROR_TYPE_ERROR, "unexpected msg %u\n",
1523              ntohs(msg->type));
1524     }
1525
1526     GNUNET_CONTAINER_DLL_remove (handle->th_head, handle->th_tail, th);
1527     GNUNET_free (th);
1528   }
1529
1530   if (NULL != handle->th)
1531   {
1532     GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
1533     handle->th = NULL;
1534   }
1535   if (NULL != handle->client)
1536   {
1537     GNUNET_CLIENT_disconnect (handle->client);
1538     handle->client = NULL;
1539   }
1540   if (GNUNET_SCHEDULER_NO_TASK != handle->reconnect_task)
1541   {
1542     GNUNET_SCHEDULER_cancel(handle->reconnect_task);
1543     handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1544   }
1545   GNUNET_free (handle);
1546 }
1547
1548
1549 /**
1550  * Create a new channel towards a remote peer.
1551  *
1552  * If the destination port is not open by any peer or the destination peer
1553  * does not accept the channel, #GNUNET_MESH_ChannelEndHandler will be called
1554  * for this channel.
1555  *
1556  * @param h mesh handle
1557  * @param channel_ctx client's channel context to associate with the channel
1558  * @param peer peer identity the channel should go to
1559  * @param port Port number.
1560  * @param options MeshOption flag field, with all desired option bits set to 1.
1561  *
1562  * @return handle to the channel
1563  */
1564 struct GNUNET_MESH_Channel *
1565 GNUNET_MESH_channel_create (struct GNUNET_MESH_Handle *h,
1566                             void *channel_ctx,
1567                             const struct GNUNET_PeerIdentity *peer,
1568                             uint32_t port,
1569                             enum GNUNET_MESH_ChannelOption options)
1570 {
1571   struct GNUNET_MESH_Channel *ch;
1572   struct GNUNET_MESH_ChannelMessage msg;
1573
1574   LOG (GNUNET_ERROR_TYPE_DEBUG,
1575        "Creating new channel to %s:%u\n",
1576        GNUNET_i2s (peer), port);
1577   ch = create_channel (h, 0);
1578   LOG (GNUNET_ERROR_TYPE_DEBUG, "  at %p\n", ch);
1579   LOG (GNUNET_ERROR_TYPE_DEBUG, "  number %X\n", ch->chid);
1580   ch->ctx = channel_ctx;
1581   ch->peer = GNUNET_PEER_intern (peer);
1582   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE);
1583   msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
1584   msg.channel_id = htonl (ch->chid);
1585   msg.port = htonl (port);
1586   msg.peer = *peer;
1587   msg.opt = htonl (options);
1588   ch->allow_send = 0;
1589   send_packet (h, &msg.header, ch);
1590   return ch;
1591 }
1592
1593
1594 void
1595 GNUNET_MESH_channel_destroy (struct GNUNET_MESH_Channel *channel)
1596 {
1597   struct GNUNET_MESH_Handle *h;
1598   struct GNUNET_MESH_ChannelMessage msg;
1599   struct GNUNET_MESH_TransmitHandle *th;
1600
1601   LOG (GNUNET_ERROR_TYPE_DEBUG, "Destroying channel\n");
1602   h = channel->mesh;
1603
1604   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY);
1605   msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
1606   msg.channel_id = htonl (channel->chid);
1607   memset (&msg.peer, 0, sizeof (struct GNUNET_PeerIdentity));
1608   msg.port = 0;
1609   msg.opt = 0;
1610   th = h->th_head;
1611   while (th != NULL)
1612   {
1613     struct GNUNET_MESH_TransmitHandle *aux;
1614     if (th->channel == channel)
1615     {
1616       aux = th->next;
1617       /* FIXME call the handler? */
1618       if (GNUNET_YES == th_is_payload (th))
1619         th->notify (th->notify_cls, 0, NULL);
1620       GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
1621       GNUNET_free (th);
1622       th = aux;
1623     }
1624     else
1625       th = th->next;
1626   }
1627
1628   destroy_channel (channel, GNUNET_YES);
1629   send_packet (h, &msg.header, NULL);
1630 }
1631
1632
1633 /**
1634  * Get information about a channel.
1635  *
1636  * @param channel Channel handle.
1637  * @param option Query (GNUNET_MESH_OPTION_*).
1638  * @param ... dependant on option, currently not used
1639  *
1640  * @return Union with an answer to the query.
1641  */
1642 const union GNUNET_MESH_ChannelInfo *
1643 GNUNET_MESH_channel_get_info (struct GNUNET_MESH_Channel *channel,
1644                               enum GNUNET_MESH_ChannelOption option, ...)
1645 {
1646   static int bool_flag;
1647   const union GNUNET_MESH_ChannelInfo *ret;
1648
1649   switch (option)
1650   {
1651     case GNUNET_MESH_OPTION_NOBUFFER:
1652     case GNUNET_MESH_OPTION_RELIABLE:
1653     case GNUNET_MESH_OPTION_OOORDER:
1654       if (0 != (option & channel->options))
1655         bool_flag = GNUNET_YES;
1656       else
1657         bool_flag = GNUNET_NO;
1658       ret = (const union GNUNET_MESH_ChannelInfo *) &bool_flag;
1659       break;
1660     case GNUNET_MESH_OPTION_PEER:
1661       ret = (const union GNUNET_MESH_ChannelInfo *) GNUNET_PEER_resolve2 (channel->peer);
1662       break;
1663     default:
1664       GNUNET_break (0);
1665       return NULL;
1666   }
1667
1668   return ret;
1669 }
1670
1671 struct GNUNET_MESH_TransmitHandle *
1672 GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Channel *channel, int cork,
1673                                    struct GNUNET_TIME_Relative maxdelay,
1674                                    size_t notify_size,
1675                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
1676                                    void *notify_cls)
1677 {
1678   struct GNUNET_MESH_TransmitHandle *th;
1679
1680   GNUNET_assert (NULL != channel);
1681   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH NOTIFY TRANSMIT READY\n");
1682   LOG (GNUNET_ERROR_TYPE_DEBUG, "    on channel %X\n", channel->chid);
1683   LOG (GNUNET_ERROR_TYPE_DEBUG, "    allow_send %d\n", channel->allow_send);
1684   if (channel->chid >= GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
1685     LOG (GNUNET_ERROR_TYPE_DEBUG, "    to origin\n");
1686   else
1687     LOG (GNUNET_ERROR_TYPE_DEBUG, "    to destination\n");
1688   LOG (GNUNET_ERROR_TYPE_DEBUG, "    payload size %u\n", notify_size);
1689   GNUNET_assert (NULL != notify);
1690   GNUNET_assert (0 == channel->packet_size); // Only one data packet allowed
1691   th = GNUNET_new (struct GNUNET_MESH_TransmitHandle);
1692   th->channel = channel;
1693   th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
1694   th->size = notify_size + sizeof (struct GNUNET_MESH_LocalData);
1695   channel->packet_size = th->size;
1696   LOG (GNUNET_ERROR_TYPE_DEBUG, "    total size %u\n", th->size);
1697   th->notify = notify;
1698   th->notify_cls = notify_cls;
1699   add_to_queue (channel->mesh, th);
1700   if (NULL != channel->mesh->th)
1701     return th;
1702   if (GNUNET_NO == channel->allow_send)
1703     return th;
1704   LOG (GNUNET_ERROR_TYPE_DEBUG, "    call client notify tmt rdy\n");
1705   channel->mesh->th =
1706       GNUNET_CLIENT_notify_transmit_ready (channel->mesh->client, th->size,
1707                                            GNUNET_TIME_UNIT_FOREVER_REL,
1708                                            GNUNET_YES, &send_callback,
1709                                            channel->mesh);
1710   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH NOTIFY TRANSMIT READY END\n");
1711   return th;
1712 }
1713
1714
1715 void
1716 GNUNET_MESH_notify_transmit_ready_cancel (struct GNUNET_MESH_TransmitHandle *th)
1717 {
1718   struct GNUNET_MESH_Handle *mesh;
1719
1720   th->channel->packet_size = 0;
1721   mesh = th->channel->mesh;
1722   if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1723     GNUNET_SCHEDULER_cancel (th->timeout_task);
1724   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
1725   GNUNET_free (th);
1726   if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
1727   {
1728     /* queue empty, no point in asking for transmission */
1729     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
1730     mesh->th = NULL;
1731   }
1732 }
1733
1734
1735 void
1736 GNUNET_MESH_receive_done (struct GNUNET_MESH_Channel *channel)
1737 {
1738   send_ack (channel);
1739 }
1740
1741
1742 static void
1743 send_info_request (struct GNUNET_MESH_Handle *h, uint16_t type)
1744 {
1745   struct GNUNET_MessageHeader msg;
1746
1747   msg.size = htons (sizeof (msg));
1748   msg.type = htons (type);
1749   send_packet (h, &msg, NULL);
1750 }
1751
1752 /**
1753  * Request information about the running mesh peer.
1754  * The callback will be called for every channel known to the service,
1755  * listing all active peers that blong to the channel.
1756  *
1757  * If called again on the same handle, it will overwrite the previous
1758  * callback and cls. To retrieve the cls, monitor_cancel must be
1759  * called first.
1760  *
1761  * WARNING: unstable API, likely to change in the future!
1762  *
1763  * @param h Handle to the mesh peer.
1764  * @param callback Function to call with the requested data.
1765  * @param callback_cls Closure for @c callback.
1766  */
1767 void
1768 GNUNET_MESH_get_channels (struct GNUNET_MESH_Handle *h,
1769                          GNUNET_MESH_ChannelsCB callback,
1770                          void *callback_cls)
1771 {
1772   send_info_request (h, GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS);
1773   h->channels_cb = callback;
1774   h->channels_cls = callback_cls;
1775 }
1776
1777
1778 /**
1779  * Cancel a monitor request. The monitor callback will not be called.
1780  *
1781  * WARNING: unstable API, likely to change in the future!
1782  *
1783  * @param h Mesh handle.
1784  *
1785  * @return Closure given to GNUNET_MESH_monitor, if any.
1786  */
1787 void *
1788 GNUNET_MESH_get_channels_cancel (struct GNUNET_MESH_Handle *h)
1789 {
1790   void *cls;
1791
1792   cls = h->channels_cls;
1793   h->channels_cb = NULL;
1794   h->channels_cls = NULL;
1795   return cls;
1796 }
1797
1798
1799 /**
1800  * Request information about the running mesh peer.
1801  * The callback will be called for every peer known to the service.
1802  *
1803  * If called again on the same handle, it will overwrite the previous
1804  * callback and cls. To retrieve the cls, monitor_cancel must be
1805  * called first.
1806  *
1807  * WARNING: unstable API, likely to change in the future!
1808  *
1809  * @param h Handle to the mesh peer.
1810  * @param callback Function to call with the requested data.
1811  * @param callback_cls Closure for @c callback.
1812  */
1813 void
1814 GNUNET_MESH_get_peers (struct GNUNET_MESH_Handle *h,
1815                        GNUNET_MESH_PeersCB callback,
1816                        void *callback_cls)
1817 {
1818   send_info_request (h, GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_PEERS);
1819   h->peers_cb = callback;
1820   h->peers_cls = callback_cls;
1821 }
1822
1823
1824
1825 /**
1826  * Request information about the running mesh peer.
1827  * The callback will be called for every tunnel known to the service.
1828  *
1829  * If called again on the same handle, it will overwrite the previous
1830  * callback and cls. To retrieve the cls, monitor_cancel must be
1831  * called first.
1832  *
1833  * WARNING: unstable API, likely to change in the future!
1834  *
1835  * @param h Handle to the mesh peer.
1836  * @param callback Function to call with the requested data.
1837  * @param callback_cls Closure for @c callback.
1838  */
1839 void
1840 GNUNET_MESH_get_tunnels (struct GNUNET_MESH_Handle *h,
1841                          GNUNET_MESH_TunnelsCB callback,
1842                          void *callback_cls)
1843 {
1844   send_info_request (h, GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNELS);
1845   h->tunnels_cb = callback;
1846   h->tunnels_cls = callback_cls;
1847 }
1848
1849
1850 /**
1851  * Cancel a monitor request. The monitor callback will not be called.
1852  *
1853  * @param h Mesh handle.
1854  *
1855  * @return Closure given to GNUNET_MESH_monitor, if any.
1856  */
1857 void *
1858 GNUNET_MESH_get_tunnels_cancel (struct GNUNET_MESH_Handle *h)
1859 {
1860   void *cls;
1861
1862   h->tunnels_cb = NULL;
1863   cls = h->tunnels_cls;
1864   h->tunnels_cls = NULL;
1865
1866   return cls;
1867 }
1868
1869
1870
1871 /**
1872  * Request information about the running mesh peer.
1873  * The callback will be called for every channel known to the service,
1874  * listing all active peers that blong to the channel.
1875  *
1876  * If called again on the same handle, it will overwrite the previous
1877  * callback and cls. To retrieve the cls, monitor_cancel must be
1878  * called first.
1879  *
1880  * WARNING: unstable API, likely to change in the future!
1881  *
1882  * @param h Handle to the mesh peer.
1883  * @param callback Function to call with the requested data.
1884  * @param callback_cls Closure for @c callback.
1885  */
1886 void
1887 GNUNET_MESH_get_tunnel (struct GNUNET_MESH_Handle *h,
1888                         const struct GNUNET_PeerIdentity *id,
1889                         GNUNET_MESH_TunnelCB callback,
1890                         void *callback_cls)
1891 {
1892   struct GNUNET_MESH_LocalInfo msg;
1893
1894   memset (&msg, 0, sizeof (msg));
1895   msg.header.size = htons (sizeof (msg));
1896   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_TUNNEL);
1897   msg.peer = *id;
1898   send_packet (h, &msg.header, NULL);
1899   h->tunnel_cb = callback;
1900   h->tunnel_cls = callback_cls;
1901 }
1902
1903
1904 /**
1905  * Request information about a specific channel of the running mesh peer.
1906  *
1907  * WARNING: unstable API, likely to change in the future!
1908  * FIXME Add destination option.
1909  *
1910  * @param h Handle to the mesh peer.
1911  * @param initiator ID of the owner of the channel.
1912  * @param channel_number Channel number.
1913  * @param callback Function to call with the requested data.
1914  * @param callback_cls Closure for @c callback.
1915  */
1916 void
1917 GNUNET_MESH_show_channel (struct GNUNET_MESH_Handle *h,
1918                          struct GNUNET_PeerIdentity *initiator,
1919                          unsigned int channel_number,
1920                          GNUNET_MESH_ChannelCB callback,
1921                          void *callback_cls)
1922 {
1923   struct GNUNET_MESH_LocalInfo msg;
1924
1925   msg.header.size = htons (sizeof (msg));
1926   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL);
1927   msg.peer = *initiator;
1928   msg.channel_id = htonl (channel_number);
1929 //   msg.reserved = 0;
1930   send_packet (h, &msg.header, NULL);
1931   h->channel_cb = callback;
1932   h->channel_cls = callback_cls;
1933 }
1934
1935
1936 /**
1937  * Function called to notify a client about the connection
1938  * begin ready to queue more data.  "buf" will be
1939  * NULL and "size" zero if the connection was closed for
1940  * writing in the meantime.
1941  *
1942  * @param cls closure
1943  * @param size number of bytes available in buf
1944  * @param buf where the callee should write the message
1945  * @return number of bytes written to buf
1946  */
1947 static size_t
1948 mesh_mq_ntr (void *cls, size_t size,
1949              void *buf)
1950 {
1951   struct GNUNET_MQ_Handle *mq = cls;
1952   struct MeshMQState *state = GNUNET_MQ_impl_state (mq);
1953   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
1954   uint16_t msize;
1955
1956   state->th = NULL;
1957   if (NULL == buf)
1958   {
1959     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_WRITE);
1960     return 0;
1961   }
1962   msize = ntohs (msg->size);
1963   GNUNET_assert (msize <= size);
1964   memcpy (buf, msg, msize);
1965   GNUNET_MQ_impl_send_continue (mq);
1966   return msize;
1967 }
1968
1969
1970 /**
1971  * Signature of functions implementing the
1972  * sending functionality of a message queue.
1973  *
1974  * @param mq the message queue
1975  * @param msg the message to send
1976  * @param impl_state state of the implementation
1977  */
1978 static void
1979 mesh_mq_send_impl (struct GNUNET_MQ_Handle *mq,
1980                    const struct GNUNET_MessageHeader *msg, void *impl_state)
1981 {
1982   struct MeshMQState *state = impl_state;
1983
1984   GNUNET_assert (NULL == state->th);
1985   state->th =
1986       GNUNET_MESH_notify_transmit_ready (state->channel,
1987                                          /* FIXME: add option for corking */
1988                                          GNUNET_NO,
1989                                          GNUNET_TIME_UNIT_FOREVER_REL,
1990                                          ntohs (msg->size),
1991                                          mesh_mq_ntr, mq);
1992
1993 }
1994
1995
1996 /**
1997  * Signature of functions implementing the
1998  * destruction of a message queue.
1999  * Implementations must not free 'mq', but should
2000  * take care of 'impl_state'.
2001  *
2002  * @param mq the message queue to destroy
2003  * @param impl_state state of the implementation
2004  */
2005 static void
2006 mesh_mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
2007 {
2008   struct MeshMQState *state = impl_state;
2009
2010   if (NULL != state->th)
2011     GNUNET_MESH_notify_transmit_ready_cancel (state->th);
2012
2013   GNUNET_free (state);
2014 }
2015
2016
2017 /**
2018  * Create a message queue for a mesh channel.
2019  * The message queue can only be used to transmit messages,
2020  * not to receive them.
2021  *
2022  * @param channel the channel to create the message qeue for
2023  * @return a message queue to messages over the channel
2024  */
2025 struct GNUNET_MQ_Handle *
2026 GNUNET_MESH_mq_create (struct GNUNET_MESH_Channel *channel)
2027 {
2028   struct GNUNET_MQ_Handle *mq;
2029   struct MeshMQState *state;
2030
2031   state = GNUNET_new (struct MeshMQState);
2032   state->channel = channel;
2033
2034   mq = GNUNET_MQ_queue_for_callbacks (mesh_mq_send_impl,
2035                                       mesh_mq_destroy_impl,
2036                                       NULL, /* FIXME: cancel impl. */
2037                                       state,
2038                                       NULL, /* no msg handlers */
2039                                       NULL, /* no err handlers */
2040                                       NULL); /* no handler cls */
2041   return mq;
2042 }
2043