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
 209 /**
 210  * Description of a peer
 211  */
 212 struct GNUNET_MESH_Peer
 213 {
 214     /**
 215      * ID of the peer in short form
 216      */
 217   GNUNET_PEER_Id id;
 218
 219   /**
 220    * Channel this peer belongs to
 221    */
 222   struct GNUNET_MESH_Channel *t;
 223
 224   /**
 225    * Flag indicating whether service has informed about its connection
 226    * FIXME-BART: is this flag used? Seems dead right now...
 227    */
 228   int connected;
 229
 230 };
 231
 232
 233 /**
 234  * Opaque handle to a channel.
 235  */
 236 struct GNUNET_MESH_Channel
 237 {
 238
 239     /**
 240      * DLL next
 241      */
 242   struct GNUNET_MESH_Channel *next;
 243
 244     /**
 245      * DLL prev
 246      */
 247   struct GNUNET_MESH_Channel *prev;
 248
 249     /**
 250      * Handle to the mesh this channel belongs to
 251      */
 252   struct GNUNET_MESH_Handle *mesh;
 253
 254     /**
 255      * Local ID of the channel
 256      */
 257   MESH_ChannelNumber chid;
 258
 259     /**
 260      * Port number.
 261      */
 262   uint32_t port;
 263
 264     /**
 265      * Other end of the channel.
 266      */
 267   GNUNET_PEER_Id peer;
 268
 269   /**
 270    * Any data the caller wants to put in here
 271    */
 272   void *ctx;
 273
 274     /**
 275      * Size of packet queued in this channel
 276      */
 277   unsigned int packet_size;
 278
 279     /**
 280      * Channel options: reliability, etc.
 281      */
 282   enum MeshOption options;
 283
 284     /**
 285      * Are we allowed to send to the service?
 286      */
 287   int allow_send;
 288
 289 };
 290
 291
 292 /**
 293  * Implementation state for mesh's message queue.
 294  */
 295 struct MeshMQState
 296 {
 297   /**
 298    * The current transmit handle, or NULL
 299    * if no transmit is active.
 300    */
 301   struct GNUNET_MESH_TransmitHandle *th;
 302
 303   /**
 304    * Channel to send the data over.
 305    */
 306   struct GNUNET_MESH_Channel *channel;
 307 };
 308
 309
 310 /******************************************************************************/
 311 /***********************         DECLARATIONS         *************************/
 312 /******************************************************************************/
 313
 314 /**
 315  * Function called to send a message to the service.
 316  * "buf" will be NULL and "size" zero if the socket was closed for writing in
 317  * the meantime.
 318  *
 319  * @param cls closure, the mesh handle
 320  * @param size number of bytes available in buf
 321  * @param buf where the callee should write the connect message
 322  * @return number of bytes written to buf
 323  */
 324 static size_t
 325 send_callback (void *cls, size_t size, void *buf);
 326
 327
 328 /******************************************************************************/
 329 /***********************     AUXILIARY FUNCTIONS      *************************/
 330 /******************************************************************************/
 331
 332 /**
 333  * Check if transmission is a payload packet.
 334  *
 335  * @param th Transmission handle.
 336  *
 337  * @return GNUNET_YES if it is a payload packet,
 338  *         GNUNET_NO if it is a mesh management packet.
 339  */
 340 static int
 341 th_is_payload (struct GNUNET_MESH_TransmitHandle *th)
 342 {
 343   return (th->notify != NULL) ? GNUNET_YES : GNUNET_NO;
 344 }
 345
 346
 347 /**
 348  * Check whether there is any message ready in the queue and find the size.
 349  *
 350  * @param h Mesh handle.
 351  *
 352  * @return The size of the first ready message in the queue,
 353  *         0 if there is none.
 354  */
 355 static size_t
 356 message_ready_size (struct GNUNET_MESH_Handle *h)
 357 {
 358   struct GNUNET_MESH_TransmitHandle *th;
 359   struct GNUNET_MESH_Channel *ch;
 360
 361   for (th = h->th_head; NULL != th; th = th->next)
 362   {
 363     ch = th->channel;
 364     if (GNUNET_NO == th_is_payload (th))
 365     {
 366       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  message internal\n");
 367       return th->size;
 368     }
 369     if (GNUNET_YES == ch->allow_send)
 370     {
 371       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  message payload ok\n");
 372       return th->size;
 373     }
 374   }
 375   return 0;
 376 }
 377
 378
 379 /**
 380  * Get the channel handler for the channel specified by id from the given handle
 381  * @param h Mesh handle
 382  * @param chid ID of the wanted channel
 383  * @return handle to the required channel or NULL if not found
 384  */
 385 static struct GNUNET_MESH_Channel *
 386 retrieve_channel (struct GNUNET_MESH_Handle *h, MESH_ChannelNumber chid)
 387 {
 388   struct GNUNET_MESH_Channel *ch;
 389
 390   ch = h->channels_head;
 391   while (ch != NULL)
 392   {
 393     if (ch->chid == chid)
 394       return ch;
 395     ch = ch->next;
 396   }
 397   return NULL;
 398 }
 399
 400
 401 /**
 402  * Create a new channel and insert it in the channel list of the mesh handle
 403  *
 404  * @param h Mesh handle
 405  * @param chid Desired chid of the channel, 0 to assign one automatically.
 406  *
 407  * @return Handle to the created channel.
 408  */
 409 static struct GNUNET_MESH_Channel *
 410 create_channel (struct GNUNET_MESH_Handle *h, MESH_ChannelNumber chid)
 411 {
 412   struct GNUNET_MESH_Channel *ch;
 413
 414   ch = GNUNET_malloc (sizeof (struct GNUNET_MESH_Channel));
 415   GNUNET_CONTAINER_DLL_insert (h->channels_head, h->channels_tail, ch);
 416   ch->mesh = h;
 417   if (0 == chid)
 418   {
 419     ch->chid = h->next_chid;
 420     while (NULL != retrieve_channel (h, h->next_chid))
 421     {
 422       h->next_chid++;
 423       h->next_chid &= ~GNUNET_MESH_LOCAL_CHANNEL_ID_SERV;
 424       h->next_chid |= GNUNET_MESH_LOCAL_CHANNEL_ID_CLI;
 425     }
 426   }
 427   else
 428   {
 429     ch->chid = chid;
 430   }
 431   ch->allow_send = GNUNET_NO;
 432   return ch;
 433 }
 434
 435
 436 /**
 437  * Destroy the specified channel.
 438  * - Destroys all peers, calling the disconnect callback on each if needed
 439  * - Cancels all outgoing traffic for that channel, calling respective notifys
 440  * - Calls cleaner if channel was inbound
 441  * - Frees all memory used
 442  *
 443  * @param ch Pointer to the channel.
 444  * @param call_cleaner Whether to call the cleaner handler.
 445  *
 446  * @return Handle to the required channel or NULL if not found.
 447  */
 448 static void
 449 destroy_channel (struct GNUNET_MESH_Channel *ch, int call_cleaner)
 450 {
 451   struct GNUNET_MESH_Handle *h;
 452   struct GNUNET_MESH_TransmitHandle *th;
 453   struct GNUNET_MESH_TransmitHandle *next;
 454
 455   LOG (GNUNET_ERROR_TYPE_DEBUG, " destroy_channel %X\n", ch->chid);
 456
 457   if (NULL == ch)
 458   {
 459     GNUNET_break (0);
 460     return;
 461   }
 462   h = ch->mesh;
 463
 464   GNUNET_CONTAINER_DLL_remove (h->channels_head, h->channels_tail, ch);
 465
 466   /* signal channel destruction */
 467   if ( (NULL != h->cleaner) && (0 != ch->peer) && (GNUNET_YES == call_cleaner) )
 468   {
 469     LOG (GNUNET_ERROR_TYPE_DEBUG, " calling cleaner\n");
 470     h->cleaner (h->cls, ch, ch->ctx);
 471   }
 472
 473   /* check that clients did not leave messages behind in the queue */
 474   for (th = h->th_head; NULL != th; th = next)
 475   {
 476     next = th->next;
 477     if (th->channel != ch)
 478       continue;
 479     /* Clients should have aborted their requests already.
 480      * Management traffic should be ok, as clients can't cancel that.
 481      * If the service crashed and we are reconnecting, it's ok.
 482      */
 483     GNUNET_break (GNUNET_NO == th_is_payload (th)
 484                   || GNUNET_NO == h->in_receive);
 485     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
 486
 487     /* clean up request */
 488     if (GNUNET_SCHEDULER_NO_TASK != th->timeout_task)
 489       GNUNET_SCHEDULER_cancel (th->timeout_task);
 490     GNUNET_free (th);
 491   }
 492
 493   /* if there are no more pending requests with mesh service, cancel active request */
 494   /* Note: this should be unnecessary... */
 495   if ((0 == message_ready_size (h)) && (NULL != h->th))
 496   {
 497     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 498     h->th = NULL;
 499   }
 500
 501   if (0 != ch->peer)
 502     GNUNET_PEER_change_rc (ch->peer, -1);
 503   GNUNET_free (ch);
 504   return;
 505 }
 506
 507
 508 /**
 509  * Notify client that the transmission has timed out
 510  *
 511  * @param cls closure
 512  * @param tc task context
 513  */
 514 static void
 515 timeout_transmission (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 516 {
 517   struct GNUNET_MESH_TransmitHandle *th = cls;
 518   struct GNUNET_MESH_Handle *mesh;
 519
 520   mesh = th->channel->mesh;
 521   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
 522   th->channel->packet_size = 0;
 523   if (GNUNET_YES == th_is_payload (th))
 524     th->notify (th->notify_cls, 0, NULL);
 525   GNUNET_free (th);
 526   if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
 527   {
 528     /* nothing ready to transmit, no point in asking for transmission */
 529     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
 530     mesh->th = NULL;
 531   }
 532 }
 533
 534
 535 /**
 536  * Add a transmit handle to the transmission queue and set the
 537  * timeout if needed.
 538  *
 539  * @param h mesh handle with the queue head and tail
 540  * @param th handle to the packet to be transmitted
 541  */
 542 static void
 543 add_to_queue (struct GNUNET_MESH_Handle *h,
 544               struct GNUNET_MESH_TransmitHandle *th)
 545 {
 546   GNUNET_CONTAINER_DLL_insert_tail (h->th_head, h->th_tail, th);
 547   if (GNUNET_TIME_UNIT_FOREVER_ABS.abs_value_us == th->timeout.abs_value_us)
 548     return;
 549   th->timeout_task =
 550       GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_absolute_get_remaining
 551                                     (th->timeout), &timeout_transmission, th);
 552 }
 553
 554
 555 /**
 556  * Auxiliary function to send an already constructed packet to the service.
 557  * Takes care of creating a new queue element, copying the message and
 558  * calling the tmt_rdy function if necessary.
 559  *
 560  * @param h mesh handle
 561  * @param msg message to transmit
 562  * @param channel channel this send is related to (NULL if N/A)
 563  */
 564 static void
 565 send_packet (struct GNUNET_MESH_Handle *h,
 566              const struct GNUNET_MessageHeader *msg,
 567              struct GNUNET_MESH_Channel *channel);
 568
 569
 570 /**
 571  * Send an ack on the channel to confirm the processing of a message.
 572  *
 573  * @param ch Channel on which to send the ACK.
 574  */
 575 static void
 576 send_ack (struct GNUNET_MESH_Channel *ch)
 577 {
 578   struct GNUNET_MESH_LocalAck msg;
 579
 580   LOG (GNUNET_ERROR_TYPE_DEBUG, "Sending ACK on channel %X\n", ch->chid);
 581   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK);
 582   msg.header.size = htons (sizeof (msg));
 583   msg.channel_id = htonl (ch->chid);
 584
 585   send_packet (ch->mesh, &msg.header, ch);
 586   return;
 587 }
 588
 589
 590
 591 /**
 592  * Reconnect callback: tries to reconnect again after a failer previous
 593  * reconnecttion
 594  * @param cls closure (mesh handle)
 595  * @param tc task context
 596  */
 597 static void
 598 reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc);
 599
 600
 601 /**
 602  * Send a connect packet to the service with the applications and types
 603  * requested by the user.
 604  *
 605  * @param h The mesh handle.
 606  *
 607  */
 608 static void
 609 send_connect (struct GNUNET_MESH_Handle *h)
 610 {
 611   size_t size;
 612
 613   size = sizeof (struct GNUNET_MESH_ClientConnect);
 614   size += h->n_ports * sizeof (uint32_t);
 615   {
 616     char buf[size] GNUNET_ALIGN;
 617     struct GNUNET_MESH_ClientConnect *msg;
 618     uint32_t *ports;
 619     uint16_t i;
 620
 621     /* build connection packet */
 622     msg = (struct GNUNET_MESH_ClientConnect *) buf;
 623     msg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT);
 624     msg->header.size = htons (size);
 625     ports = (uint32_t *) &msg[1];
 626     for (i = 0; i < h->n_ports; i++)
 627     {
 628       ports[i] = htonl (h->ports[i]);
 629       LOG (GNUNET_ERROR_TYPE_DEBUG, " port %u\n",
 630            h->ports[i]);
 631     }
 632     LOG (GNUNET_ERROR_TYPE_DEBUG,
 633          "Sending %lu bytes long message with %u ports\n",
 634          ntohs (msg->header.size), h->n_ports);
 635     send_packet (h, &msg->header, NULL);
 636   }
 637 }
 638
 639
 640 /**
 641  * Reconnect to the service, retransmit all infomation to try to restore the
 642  * original state.
 643  *
 644  * @param h handle to the mesh
 645  *
 646  * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
 647  */
 648 static int
 649 do_reconnect (struct GNUNET_MESH_Handle *h)
 650 {
 651   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 652   LOG (GNUNET_ERROR_TYPE_DEBUG, "*******   RECONNECT   *******\n");
 653   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 654   LOG (GNUNET_ERROR_TYPE_DEBUG, "******** on %p *******\n", h);
 655   LOG (GNUNET_ERROR_TYPE_DEBUG, "*****************************\n");
 656
 657   /* disconnect */
 658   if (NULL != h->th)
 659   {
 660     GNUNET_CLIENT_notify_transmit_ready_cancel (h->th);
 661     h->th = NULL;
 662   }
 663   if (NULL != h->client)
 664   {
 665     GNUNET_CLIENT_disconnect (h->client);
 666   }
 667
 668   /* connect again */
 669   h->client = GNUNET_CLIENT_connect ("mesh", h->cfg);
 670   if (h->client == NULL)
 671   {
 672     h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 673                                                       &reconnect_cbk, h);
 674     h->reconnect_time =
 675         GNUNET_TIME_relative_min (GNUNET_TIME_UNIT_SECONDS,
 676                                   GNUNET_TIME_relative_multiply
 677                                   (h->reconnect_time, 2));
 678     LOG (GNUNET_ERROR_TYPE_DEBUG, "Next retry in %s\n",
 679          GNUNET_STRINGS_relative_time_to_string (h->reconnect_time,
 680                                                  GNUNET_NO));
 681     GNUNET_break (0);
 682     return GNUNET_NO;
 683   }
 684   else
 685   {
 686     h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
 687   }
 688   send_connect (h);
 689   return GNUNET_YES;
 690 }
 691
 692 /**
 693  * Reconnect callback: tries to reconnect again after a failer previous
 694  * reconnecttion
 695  * @param cls closure (mesh handle)
 696  * @param tc task context
 697  */
 698 static void
 699 reconnect_cbk (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 700 {
 701   struct GNUNET_MESH_Handle *h = cls;
 702
 703   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
 704   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
 705     return;
 706   do_reconnect (h);
 707 }
 708
 709
 710 /**
 711  * Reconnect to the service, retransmit all infomation to try to restore the
 712  * original state.
 713  *
 714  * @param h handle to the mesh
 715  *
 716  * @return GNUNET_YES in case of sucess, GNUNET_NO otherwise (service down...)
 717  */
 718 static void
 719 reconnect (struct GNUNET_MESH_Handle *h)
 720 {
 721   struct GNUNET_MESH_Channel *ch;
 722   struct GNUNET_MESH_Channel *next;
 723
 724   LOG (GNUNET_ERROR_TYPE_DEBUG,
 725        "Requested RECONNECT, destroying all channels\n");
 726   h->in_receive = GNUNET_NO;
 727   for (ch = h->channels_head; NULL != ch; ch = next)
 728   {
 729     next = ch->next;
 730     destroy_channel (ch, GNUNET_YES);
 731   }
 732   if (GNUNET_SCHEDULER_NO_TASK == h->reconnect_task)
 733     h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 734                                                       &reconnect_cbk, h);
 735 }
 736
 737
 738 /******************************************************************************/
 739 /***********************      RECEIVE HANDLERS     ****************************/
 740 /******************************************************************************/
 741
 742 /**
 743  * Process the new channel notification and add it to the channels in the handle
 744  *
 745  * @param h     The mesh handle
 746  * @param msg   A message with the details of the new incoming channel
 747  */
 748 static void
 749 process_channel_created (struct GNUNET_MESH_Handle *h,
 750                         const struct GNUNET_MESH_ChannelMessage *msg)
 751 {
 752   struct GNUNET_MESH_Channel *ch;
 753   MESH_ChannelNumber chid;
 754   uint32_t port;
 755
 756   chid = ntohl (msg->channel_id);
 757   port = ntohl (msg->port);
 758   LOG (GNUNET_ERROR_TYPE_DEBUG, "Creating incoming channel %X:%u\n", chid, port);
 759   if (chid < GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
 760   {
 761     GNUNET_break (0);
 762     return;
 763   }
 764   if (NULL != h->new_channel)
 765   {
 766     void *ctx;
 767
 768     ch = create_channel (h, chid);
 769     ch->allow_send = GNUNET_NO;
 770     ch->peer = GNUNET_PEER_intern (&msg->peer);
 771     ch->mesh = h;
 772     ch->chid = chid;
 773     ch->port = port;
 774     ch->options = ntohl (msg->opt);
 775
 776     LOG (GNUNET_ERROR_TYPE_DEBUG, "  created channel %p\n", ch);
 777     ctx = h->new_channel (h->cls, ch, &msg->peer, ch->port, ch->options);
 778     if (NULL != ctx)
 779       ch->ctx = ctx;
 780     LOG (GNUNET_ERROR_TYPE_DEBUG, "User notified\n");
 781   }
 782   else
 783   {
 784     struct GNUNET_MESH_ChannelMessage d_msg;
 785
 786     LOG (GNUNET_ERROR_TYPE_DEBUG, "No handler for incoming channels\n");
 787
 788     d_msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY);
 789     d_msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
 790     d_msg.channel_id = msg->channel_id;
 791     memset (&d_msg.peer, 0, sizeof (struct GNUNET_PeerIdentity));
 792     d_msg.port = 0;
 793     d_msg.opt = 0;
 794
 795     send_packet (h, &d_msg.header, NULL);
 796   }
 797   return;
 798 }
 799
 800
 801 /**
 802  * Process the channel destroy notification and free associated resources
 803  *
 804  * @param h     The mesh handle
 805  * @param msg   A message with the details of the channel being destroyed
 806  */
 807 static void
 808 process_channel_destroy (struct GNUNET_MESH_Handle *h,
 809                          const struct GNUNET_MESH_ChannelMessage *msg)
 810 {
 811   struct GNUNET_MESH_Channel *ch;
 812   MESH_ChannelNumber chid;
 813
 814   LOG (GNUNET_ERROR_TYPE_DEBUG, "Channel Destroy received from service\n");
 815   chid = ntohl (msg->channel_id);
 816   ch = retrieve_channel (h, chid);
 817
 818   if (NULL == ch)
 819   {
 820     LOG (GNUNET_ERROR_TYPE_DEBUG, "channel %X unknown\n", chid);
 821     return;
 822   }
 823   LOG (GNUNET_ERROR_TYPE_DEBUG, " destroying channel %X\n", ch->chid);
 824   destroy_channel (ch, GNUNET_YES);
 825 }
 826
 827
 828 /**
 829  * Process the incoming data packets, call appropriate handlers.
 830  *
 831  * @param h         The mesh handle
 832  * @param message   A message encapsulating the data
 833  */
 834 static void
 835 process_incoming_data (struct GNUNET_MESH_Handle *h,
 836                        const struct GNUNET_MessageHeader *message)
 837 {
 838   const struct GNUNET_MessageHeader *payload;
 839   const struct GNUNET_MESH_MessageHandler *handler;
 840   struct GNUNET_MESH_LocalData *dmsg;
 841   struct GNUNET_MESH_Channel *ch;
 842   unsigned int i;
 843   uint16_t type;
 844
 845   LOG (GNUNET_ERROR_TYPE_DEBUG,
 846        "Got a data message!\n");
 847   dmsg = (struct GNUNET_MESH_LocalData *) message;
 848   ch = retrieve_channel (h, ntohl (dmsg->id));
 849   payload = (struct GNUNET_MessageHeader *) &dmsg[1];
 850   LOG (GNUNET_ERROR_TYPE_DEBUG,
 851        "  %s data on channel %s [%X]\n",
 852        ch->chid >= GNUNET_MESH_LOCAL_CHANNEL_ID_SERV ? "fwd" : "bck",
 853        GNUNET_i2s (GNUNET_PEER_resolve2 (ch->peer)), ntohl (dmsg->id));
 854   if (NULL == ch)
 855   {
 856     /* Channel was ignored/destroyed, probably service didn't get it yet */
 857     LOG (GNUNET_ERROR_TYPE_DEBUG, "  ignored!\n");
 858     return;
 859   }
 860   type = ntohs (payload->type);
 861   LOG (GNUNET_ERROR_TYPE_DEBUG, "  payload type %u\n", type);
 862   for (i = 0; i < h->n_handlers; i++)
 863   {
 864     handler = &h->message_handlers[i];
 865     LOG (GNUNET_ERROR_TYPE_DEBUG,
 866          "    checking handler for type %u\n",
 867          handler->type);
 868     if (handler->type == type)
 869     {
 870       if (GNUNET_OK !=
 871           handler->callback (h->cls, ch, &ch->ctx, payload))
 872       {
 873         LOG (GNUNET_ERROR_TYPE_DEBUG, "callback caused disconnection\n");
 874         GNUNET_MESH_channel_destroy (ch);
 875         return;
 876       }
 877       else
 878       {
 879         LOG (GNUNET_ERROR_TYPE_DEBUG,
 880              "callback completed successfully\n");
 881         return;
 882       }
 883     }
 884   }
 885 }
 886
 887
 888 /**
 889  * Process a local ACK message, enabling the client to send
 890  * more data to the service.
 891  *
 892  * @param h Mesh handle.
 893  * @param message Message itself.
 894  */
 895 static void
 896 process_ack (struct GNUNET_MESH_Handle *h,
 897              const struct GNUNET_MessageHeader *message)
 898 {
 899   struct GNUNET_MESH_LocalAck *msg;
 900   struct GNUNET_MESH_Channel *ch;
 901   MESH_ChannelNumber chid;
 902
 903   LOG (GNUNET_ERROR_TYPE_DEBUG, "Got an ACK!\n");
 904   msg = (struct GNUNET_MESH_LocalAck *) message;
 905   chid = ntohl (msg->channel_id);
 906     ch = retrieve_channel (h, chid);
 907   if (NULL == ch)
 908   {
 909     LOG (GNUNET_ERROR_TYPE_WARNING, "ACK on unknown channel %X\n", chid);
 910     return;
 911   }
 912   LOG (GNUNET_ERROR_TYPE_DEBUG, "  on channel %X!\n", ch->chid);
 913   ch->allow_send = GNUNET_YES;
 914   if (NULL == h->th && 0 < ch->packet_size)
 915   {
 916     LOG (GNUNET_ERROR_TYPE_DEBUG, "  tmt rdy was NULL, requesting!\n");
 917     h->th =
 918         GNUNET_CLIENT_notify_transmit_ready (h->client, ch->packet_size,
 919                                              GNUNET_TIME_UNIT_FOREVER_REL,
 920                                              GNUNET_YES, &send_callback, h);
 921   }
 922 }
 923
 924
 925 /*
 926  * Process a local reply about info on all channels, pass info to the user.
 927  *
 928  * @param h Mesh handle.
 929  * @param message Message itself.
 930  */
 931 // static void
 932 // process_get_channels (struct GNUNET_MESH_Handle *h,
 933 //                      const struct GNUNET_MessageHeader *message)
 934 // {
 935 //   struct GNUNET_MESH_LocalMonitor *msg;
 936 //
 937 //   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Get Channels messasge received\n");
 938 //
 939 //   if (NULL == h->channels_cb)
 940 //   {
 941 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "  ignored\n");
 942 //     return;
 943 //   }
 944 //
 945 //   msg = (struct GNUNET_MESH_LocalMonitor *) message;
 946 //   if (ntohs (message->size) !=
 947 //       (sizeof (struct GNUNET_MESH_LocalMonitor) +
 948 //        sizeof (struct GNUNET_PeerIdentity)))
 949 //   {
 950 //     GNUNET_break_op (0);
 951 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 952 //                 "Get channels message: size %hu - expected %u\n",
 953 //                 ntohs (message->size),
 954 //                 sizeof (struct GNUNET_MESH_LocalMonitor));
 955 //     return;
 956 //   }
 957 //   h->channels_cb (h->channels_cls,
 958 //                   ntohl (msg->channel_id),
 959 //                   &msg->owner,
 960 //                   &msg->destination);
 961 // }
 962
 963
 964
 965 /*
 966  * Process a local monitor_channel reply, pass info to the user.
 967  *
 968  * @param h Mesh handle.
 969  * @param message Message itself.
 970  */
 971 // static void
 972 // process_show_channel (struct GNUNET_MESH_Handle *h,
 973 //                      const struct GNUNET_MessageHeader *message)
 974 // {
 975 //   struct GNUNET_MESH_LocalMonitor *msg;
 976 //   size_t esize;
 977 //
 978 //   GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Show Channel messasge received\n");
 979 //
 980 //   if (NULL == h->channel_cb)
 981 //   {
 982 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "  ignored\n");
 983 //     return;
 984 //   }
 985 //
 986 //   /* Verify message sanity */
 987 //   msg = (struct GNUNET_MESH_LocalMonitor *) message;
 988 //   esize = sizeof (struct GNUNET_MESH_LocalMonitor);
 989 //   if (ntohs (message->size) != esize)
 990 //   {
 991 //     GNUNET_break_op (0);
 992 //     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 993 //                 "Show channel message: size %hu - expected %u\n",
 994 //                 ntohs (message->size),
 995 //                 esize);
 996 //
 997 //     h->channel_cb (h->channel_cls, NULL, NULL);
 998 //     h->channel_cb = NULL;
 999 //     h->channel_cls = NULL;
1000 //
1001 //     return;
1002 //   }
1003 //
1004 //   h->channel_cb (h->channel_cls,
1005 //                  &msg->destination,
1006 //                  &msg->owner);
1007 // }
1008
1009
1010 /**
1011  * Function to process all messages received from the service
1012  *
1013  * @param cls closure
1014  * @param msg message received, NULL on timeout or fatal error
1015  */
1016 static void
1017 msg_received (void *cls, const struct GNUNET_MessageHeader *msg)
1018 {
1019   struct GNUNET_MESH_Handle *h = cls;
1020   uint16_t type;
1021
1022   if (msg == NULL)
1023   {
1024     LOG (GNUNET_ERROR_TYPE_DEBUG,
1025          "Mesh service disconnected, reconnecting\n", h);
1026     reconnect (h);
1027     return;
1028   }
1029   type = ntohs (msg->type);
1030   LOG (GNUNET_ERROR_TYPE_DEBUG, "\n");
1031   LOG (GNUNET_ERROR_TYPE_DEBUG, "Received a message: %s\n",
1032        GM_m2s (type));
1033   switch (type)
1034   {
1035     /* Notify of a new incoming channel */
1036   case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
1037     process_channel_created (h, (struct GNUNET_MESH_ChannelMessage *) msg);
1038     break;
1039     /* Notify of a channel disconnection */
1040   case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY: /* TODO separate(gid problem)*/
1041   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_NACK:
1042     process_channel_destroy (h, (struct GNUNET_MESH_ChannelMessage *) msg);
1043     break;
1044   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_DATA:
1045     process_incoming_data (h, msg);
1046     break;
1047   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_ACK:
1048     process_ack (h, msg);
1049     break;
1050 //   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS: DEPRECATED
1051 //     process_get_channels (h, msg);
1052 //     break;
1053 //   case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL: DEPRECATED
1054 //     process_show_channel (h, msg);
1055 //     break;
1056   default:
1057     /* We shouldn't get any other packages, log and ignore */
1058     LOG (GNUNET_ERROR_TYPE_WARNING,
1059          "unsolicited message form service (type %s)\n",
1060          GM_m2s (ntohs (msg->type)));
1061   }
1062   LOG (GNUNET_ERROR_TYPE_DEBUG, "message processed\n");
1063   if (GNUNET_YES == h->in_receive)
1064   {
1065     GNUNET_CLIENT_receive (h->client, &msg_received, h,
1066                            GNUNET_TIME_UNIT_FOREVER_REL);
1067   }
1068   else
1069   {
1070     LOG (GNUNET_ERROR_TYPE_DEBUG,
1071          "in receive off, not calling CLIENT_receive\n");
1072   }
1073 }
1074
1075
1076 /******************************************************************************/
1077 /************************       SEND FUNCTIONS     ****************************/
1078 /******************************************************************************/
1079
1080 /**
1081  * Function called to send a message to the service.
1082  * "buf" will be NULL and "size" zero if the socket was closed for writing in
1083  * the meantime.
1084  *
1085  * @param cls closure, the mesh handle
1086  * @param size number of bytes available in buf
1087  * @param buf where the callee should write the connect message
1088  * @return number of bytes written to buf
1089  */
1090 static size_t
1091 send_callback (void *cls, size_t size, void *buf)
1092 {
1093   struct GNUNET_MESH_Handle *h = cls;
1094   struct GNUNET_MESH_TransmitHandle *th;
1095   struct GNUNET_MESH_TransmitHandle *next;
1096   struct GNUNET_MESH_Channel *ch;
1097   char *cbuf = buf;
1098   size_t tsize;
1099   size_t psize;
1100   size_t nsize;
1101
1102   LOG (GNUNET_ERROR_TYPE_DEBUG, "\n");
1103   LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() Buffer %u\n", size);
1104   if ((0 == size) || (NULL == buf))
1105   {
1106     LOG (GNUNET_ERROR_TYPE_DEBUG, "# Received NULL send callback on %p\n", h);
1107     reconnect (h);
1108     h->th = NULL;
1109     return 0;
1110   }
1111   tsize = 0;
1112   next = h->th_head;
1113   nsize = message_ready_size (h);
1114   while ((NULL != (th = next)) && (0 < nsize) && (size >= nsize))
1115   {
1116     ch = th->channel;
1117     if (GNUNET_YES == th_is_payload (th))
1118     {
1119       struct GNUNET_MESH_LocalData *dmsg;
1120       struct GNUNET_MessageHeader *mh;
1121
1122       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  payload\n");
1123       if (GNUNET_NO == ch->allow_send)
1124       {
1125         /* This channel is not ready to transmit yet, try next message */
1126         next = th->next;
1127         continue;
1128       }
1129       ch->packet_size = 0;
1130       GNUNET_assert (size >= th->size);
1131       dmsg = (struct GNUNET_MESH_LocalData *) cbuf;
1132       mh = (struct GNUNET_MessageHeader *) &dmsg[1];
1133       psize = th->notify (th->notify_cls,
1134                           size - sizeof (struct GNUNET_MESH_LocalData),
1135                           mh);
1136       if (psize > 0)
1137       {
1138         psize += sizeof (struct GNUNET_MESH_LocalData);
1139         GNUNET_assert (size >= psize);
1140         dmsg->header.size = htons (psize);
1141         dmsg->id = htonl (ch->chid);
1142         dmsg->header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_DATA);
1143         LOG (GNUNET_ERROR_TYPE_DEBUG, "#  payload type %s\n",
1144              GM_m2s (ntohs (mh->type)));
1145                 ch->allow_send = GNUNET_NO;
1146       }
1147       else
1148       {
1149         LOG (GNUNET_ERROR_TYPE_DEBUG,
1150              "#  callback returned size 0, "
1151              "application canceled transmission\n");
1152       }
1153     }
1154     else
1155     {
1156       struct GNUNET_MessageHeader *mh = (struct GNUNET_MessageHeader *) &th[1];
1157
1158       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  mesh internal traffic, type %s\n",
1159            GM_m2s (ntohs (mh->type)));
1160       memcpy (cbuf, &th[1], th->size);
1161       psize = th->size;
1162     }
1163     if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1164       GNUNET_SCHEDULER_cancel (th->timeout_task);
1165     GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
1166     GNUNET_free (th);
1167     next = h->th_head;
1168     nsize = message_ready_size (h);
1169     cbuf += psize;
1170     size -= psize;
1171     tsize += psize;
1172   }
1173   LOG (GNUNET_ERROR_TYPE_DEBUG, "#  total size: %u\n", tsize);
1174   h->th = NULL;
1175   size = message_ready_size (h);
1176   if (0 != size)
1177   {
1178     LOG (GNUNET_ERROR_TYPE_DEBUG, "#  next size: %u\n", size);
1179     h->th =
1180         GNUNET_CLIENT_notify_transmit_ready (h->client, size,
1181                                              GNUNET_TIME_UNIT_FOREVER_REL,
1182                                              GNUNET_YES, &send_callback, h);
1183   }
1184   else
1185   {
1186     if (NULL != h->th_head)
1187       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  can't transmit any more\n");
1188     else
1189       LOG (GNUNET_ERROR_TYPE_DEBUG, "#  nothing left to transmit\n");
1190   }
1191   if (GNUNET_NO == h->in_receive)
1192   {
1193     LOG (GNUNET_ERROR_TYPE_DEBUG, "# start receiving from service\n");
1194     h->in_receive = GNUNET_YES;
1195     GNUNET_CLIENT_receive (h->client, &msg_received, h,
1196                            GNUNET_TIME_UNIT_FOREVER_REL);
1197   }
1198   LOG (GNUNET_ERROR_TYPE_DEBUG, "# Send packet() END\n");
1199   return tsize;
1200 }
1201
1202
1203 /**
1204  * Auxiliary function to send an already constructed packet to the service.
1205  * Takes care of creating a new queue element, copying the message and
1206  * calling the tmt_rdy function if necessary.
1207  *
1208  * @param h mesh handle
1209  * @param msg message to transmit
1210  * @param channel channel this send is related to (NULL if N/A)
1211  */
1212 static void
1213 send_packet (struct GNUNET_MESH_Handle *h,
1214              const struct GNUNET_MessageHeader *msg,
1215              struct GNUNET_MESH_Channel *channel)
1216 {
1217   struct GNUNET_MESH_TransmitHandle *th;
1218   size_t msize;
1219
1220   LOG (GNUNET_ERROR_TYPE_DEBUG, " Sending message to service: %s\n",
1221        GM_m2s(ntohs(msg->type)));
1222   msize = ntohs (msg->size);
1223   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle) + msize);
1224   th->timeout = GNUNET_TIME_UNIT_FOREVER_ABS;
1225   th->size = msize;
1226   th->channel = channel;
1227   memcpy (&th[1], msg, msize);
1228   add_to_queue (h, th);
1229   LOG (GNUNET_ERROR_TYPE_DEBUG, "  queued\n");
1230   if (NULL != h->th)
1231     return;
1232   LOG (GNUNET_ERROR_TYPE_DEBUG, "  calling ntfy tmt rdy for %u bytes\n", msize);
1233   h->th =
1234       GNUNET_CLIENT_notify_transmit_ready (h->client, msize,
1235                                            GNUNET_TIME_UNIT_FOREVER_REL,
1236                                            GNUNET_YES, &send_callback, h);
1237 }
1238
1239
1240 /******************************************************************************/
1241 /**********************      API CALL DEFINITIONS     *************************/
1242 /******************************************************************************/
1243
1244 struct GNUNET_MESH_Handle *
1245 GNUNET_MESH_connect (const struct GNUNET_CONFIGURATION_Handle *cfg, void *cls,
1246                      GNUNET_MESH_InboundChannelNotificationHandler new_channel,
1247                      GNUNET_MESH_ChannelEndHandler cleaner,
1248                      const struct GNUNET_MESH_MessageHandler *handlers,
1249                      const uint32_t *ports)
1250 {
1251   struct GNUNET_MESH_Handle *h;
1252
1253   LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect()\n");
1254   h = GNUNET_malloc (sizeof (struct GNUNET_MESH_Handle));
1255   LOG (GNUNET_ERROR_TYPE_DEBUG, " addr %p\n", h);
1256   h->cfg = cfg;
1257   h->new_channel = new_channel;
1258   h->cleaner = cleaner;
1259   h->client = GNUNET_CLIENT_connect ("mesh", cfg);
1260   if (h->client == NULL)
1261   {
1262     GNUNET_break (0);
1263     GNUNET_free (h);
1264     return NULL;
1265   }
1266   h->cls = cls;
1267   h->message_handlers = handlers;
1268   h->ports = ports;
1269   h->next_chid = GNUNET_MESH_LOCAL_CHANNEL_ID_CLI;
1270   h->reconnect_time = GNUNET_TIME_UNIT_MILLISECONDS;
1271   h->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1272
1273   if (NULL != ports && ports[0] != 0 && NULL == new_channel)
1274   {
1275     GNUNET_break (0);
1276     LOG (GNUNET_ERROR_TYPE_DEBUG,
1277          "no new channel handler given, ports parameter is useless!!\n");
1278   }
1279   if ((NULL == ports || ports[0] == 0) && NULL != new_channel)
1280   {
1281     GNUNET_break (0);
1282     LOG (GNUNET_ERROR_TYPE_DEBUG,
1283          "no ports given, new channel handler will never be called!!\n");
1284   }
1285   /* count handlers */
1286   for (h->n_handlers = 0;
1287        handlers && handlers[h->n_handlers].type;
1288        h->n_handlers++) ;
1289   for (h->n_ports = 0;
1290        ports && ports[h->n_ports];
1291        h->n_ports++) ;
1292   send_connect (h);
1293   LOG (GNUNET_ERROR_TYPE_DEBUG, "GNUNET_MESH_connect() END\n");
1294   return h;
1295 }
1296
1297
1298 void
1299 GNUNET_MESH_disconnect (struct GNUNET_MESH_Handle *handle)
1300 {
1301   struct GNUNET_MESH_Channel *ch;
1302   struct GNUNET_MESH_Channel *aux;
1303   struct GNUNET_MESH_TransmitHandle *th;
1304
1305   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH DISCONNECT\n");
1306
1307   ch = handle->channels_head;
1308   while (NULL != ch)
1309   {
1310     aux = ch->next;
1311     if (ch->chid < GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
1312     {
1313       GNUNET_break (0);
1314       LOG (GNUNET_ERROR_TYPE_DEBUG, "channel %X not destroyed\n", ch->chid);
1315     }
1316     destroy_channel (ch, GNUNET_YES);
1317     ch = aux;
1318   }
1319   while ( (th = handle->th_head) != NULL)
1320   {
1321     struct GNUNET_MessageHeader *msg;
1322
1323     /* Make sure it is an allowed packet (everything else should have been
1324      * already canceled).
1325      */
1326     GNUNET_break (GNUNET_NO == th_is_payload (th));
1327     msg = (struct GNUNET_MessageHeader *) &th[1];
1328     switch (ntohs(msg->type))
1329     {
1330       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_CONNECT:
1331       case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE:
1332       case GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY:
1333       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS:
1334       case GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL:
1335         break;
1336       default:
1337         GNUNET_break (0);
1338         LOG (GNUNET_ERROR_TYPE_ERROR, "unexpected msg %u\n",
1339              ntohs(msg->type));
1340     }
1341
1342     GNUNET_CONTAINER_DLL_remove (handle->th_head, handle->th_tail, th);
1343     GNUNET_free (th);
1344   }
1345
1346   if (NULL != handle->th)
1347   {
1348     GNUNET_CLIENT_notify_transmit_ready_cancel (handle->th);
1349     handle->th = NULL;
1350   }
1351   if (NULL != handle->client)
1352   {
1353     GNUNET_CLIENT_disconnect (handle->client);
1354     handle->client = NULL;
1355   }
1356   if (GNUNET_SCHEDULER_NO_TASK != handle->reconnect_task)
1357   {
1358     GNUNET_SCHEDULER_cancel(handle->reconnect_task);
1359     handle->reconnect_task = GNUNET_SCHEDULER_NO_TASK;
1360   }
1361   GNUNET_free (handle);
1362 }
1363
1364
1365 /**
1366  * Create a new channel towards a remote peer.
1367  *
1368  * If the destination port is not open by any peer or the destination peer
1369  * does not accept the channel, #GNUNET_MESH_ChannelEndHandler will be called
1370  * for this channel.
1371  *
1372  * @param h mesh handle
1373  * @param channel_ctx client's channel context to associate with the channel
1374  * @param peer peer identity the channel should go to
1375  * @param port Port number.
1376  * @param options MeshOption flag field, with all desired option bits set to 1.
1377  *
1378  * @return handle to the channel
1379  */
1380 struct GNUNET_MESH_Channel *
1381 GNUNET_MESH_channel_create (struct GNUNET_MESH_Handle *h,
1382                             void *channel_ctx,
1383                             const struct GNUNET_PeerIdentity *peer,
1384                             uint32_t port,
1385                             enum MeshOption options)
1386 {
1387   struct GNUNET_MESH_Channel *ch;
1388   struct GNUNET_MESH_ChannelMessage msg;
1389
1390   LOG (GNUNET_ERROR_TYPE_DEBUG,
1391        "Creating new channel to %s:%u\n",
1392        GNUNET_i2s (peer), port);
1393   ch = create_channel (h, 0);
1394   LOG (GNUNET_ERROR_TYPE_DEBUG, "  at %p\n", ch);
1395   LOG (GNUNET_ERROR_TYPE_DEBUG, "  number %X\n", ch->chid);
1396   ch->ctx = channel_ctx;
1397   ch->peer = GNUNET_PEER_intern (peer);
1398   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_CREATE);
1399   msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
1400   msg.channel_id = htonl (ch->chid);
1401   msg.port = htonl (port);
1402   msg.peer = *peer;
1403   msg.opt = htonl (options);
1404   ch->allow_send = 0;
1405   send_packet (h, &msg.header, ch);
1406   return ch;
1407 }
1408
1409
1410 void
1411 GNUNET_MESH_channel_destroy (struct GNUNET_MESH_Channel *channel)
1412 {
1413   struct GNUNET_MESH_Handle *h;
1414   struct GNUNET_MESH_ChannelMessage msg;
1415   struct GNUNET_MESH_TransmitHandle *th;
1416
1417   LOG (GNUNET_ERROR_TYPE_DEBUG, "Destroying channel\n");
1418   h = channel->mesh;
1419
1420   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_CHANNEL_DESTROY);
1421   msg.header.size = htons (sizeof (struct GNUNET_MESH_ChannelMessage));
1422   msg.channel_id = htonl (channel->chid);
1423   memset (&msg.peer, 0, sizeof (struct GNUNET_PeerIdentity));
1424   msg.port = 0;
1425   msg.opt = 0;
1426   th = h->th_head;
1427   while (th != NULL)
1428   {
1429     struct GNUNET_MESH_TransmitHandle *aux;
1430     if (th->channel == channel)
1431     {
1432       aux = th->next;
1433       /* FIXME call the handler? */
1434       if (GNUNET_YES == th_is_payload (th))
1435         th->notify (th->notify_cls, 0, NULL);
1436       GNUNET_CONTAINER_DLL_remove (h->th_head, h->th_tail, th);
1437       GNUNET_free (th);
1438       th = aux;
1439     }
1440     else
1441       th = th->next;
1442   }
1443
1444   destroy_channel (channel, GNUNET_YES);
1445   send_packet (h, &msg.header, NULL);
1446 }
1447
1448
1449 /**
1450  * Get information about a channel.
1451  *
1452  * @param channel Channel handle.
1453  * @param option Query (GNUNET_MESH_OPTION_*).
1454  * @param ... dependant on option, currently not used
1455  *
1456  * @return Union with an answer to the query.
1457  */
1458 const union GNUNET_MESH_ChannelInfo *
1459 GNUNET_MESH_channel_get_info (struct GNUNET_MESH_Channel *channel,
1460                               enum MeshOption option, ...)
1461 {
1462   static int bool_flag;
1463   const union GNUNET_MESH_ChannelInfo *ret;
1464
1465   switch (option)
1466   {
1467     case GNUNET_MESH_OPTION_NOBUFFER:
1468     case GNUNET_MESH_OPTION_RELIABLE:
1469     case GNUNET_MESH_OPTION_OOORDER:
1470       if (0 != (option & channel->options))
1471         bool_flag = GNUNET_YES;
1472       else
1473         bool_flag = GNUNET_NO;
1474       ret = (const union GNUNET_MESH_ChannelInfo *) &bool_flag;
1475       break;
1476     case GNUNET_MESH_OPTION_PEER:
1477       ret = (const union GNUNET_MESH_ChannelInfo *) GNUNET_PEER_resolve2 (channel->peer);
1478       break;
1479     default:
1480       GNUNET_break (0);
1481       return NULL;
1482   }
1483
1484   return ret;
1485 }
1486
1487 struct GNUNET_MESH_TransmitHandle *
1488 GNUNET_MESH_notify_transmit_ready (struct GNUNET_MESH_Channel *channel, int cork,
1489                                    struct GNUNET_TIME_Relative maxdelay,
1490                                    size_t notify_size,
1491                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
1492                                    void *notify_cls)
1493 {
1494   struct GNUNET_MESH_TransmitHandle *th;
1495
1496   GNUNET_assert (NULL != channel);
1497   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH NOTIFY TRANSMIT READY\n");
1498   LOG (GNUNET_ERROR_TYPE_DEBUG, "    on channel %X\n", channel->chid);
1499   LOG (GNUNET_ERROR_TYPE_DEBUG, "    allow_send %d\n", channel->allow_send);
1500   if (channel->chid >= GNUNET_MESH_LOCAL_CHANNEL_ID_SERV)
1501     LOG (GNUNET_ERROR_TYPE_DEBUG, "    to origin\n");
1502   else
1503     LOG (GNUNET_ERROR_TYPE_DEBUG, "    to destination\n");
1504   LOG (GNUNET_ERROR_TYPE_DEBUG, "    payload size %u\n", notify_size);
1505   GNUNET_assert (NULL != notify);
1506   GNUNET_assert (0 == channel->packet_size); // Only one data packet allowed
1507   th = GNUNET_malloc (sizeof (struct GNUNET_MESH_TransmitHandle));
1508   th->channel = channel;
1509   th->timeout = GNUNET_TIME_relative_to_absolute (maxdelay);
1510   th->size = notify_size + sizeof (struct GNUNET_MESH_LocalData);
1511   channel->packet_size = th->size;
1512   LOG (GNUNET_ERROR_TYPE_DEBUG, "    total size %u\n", th->size);
1513   th->notify = notify;
1514   th->notify_cls = notify_cls;
1515   add_to_queue (channel->mesh, th);
1516   if (NULL != channel->mesh->th)
1517     return th;
1518   if (GNUNET_NO == channel->allow_send)
1519     return th;
1520   LOG (GNUNET_ERROR_TYPE_DEBUG, "    call client notify tmt rdy\n");
1521   channel->mesh->th =
1522       GNUNET_CLIENT_notify_transmit_ready (channel->mesh->client, th->size,
1523                                            GNUNET_TIME_UNIT_FOREVER_REL,
1524                                            GNUNET_YES, &send_callback,
1525                                            channel->mesh);
1526   LOG (GNUNET_ERROR_TYPE_DEBUG, "MESH NOTIFY TRANSMIT READY END\n");
1527   return th;
1528 }
1529
1530
1531 void
1532 GNUNET_MESH_notify_transmit_ready_cancel (struct GNUNET_MESH_TransmitHandle *th)
1533 {
1534   struct GNUNET_MESH_Handle *mesh;
1535
1536   th->channel->packet_size = 0;
1537   mesh = th->channel->mesh;
1538   if (th->timeout_task != GNUNET_SCHEDULER_NO_TASK)
1539     GNUNET_SCHEDULER_cancel (th->timeout_task);
1540   GNUNET_CONTAINER_DLL_remove (mesh->th_head, mesh->th_tail, th);
1541   GNUNET_free (th);
1542   if ((0 == message_ready_size (mesh)) && (NULL != mesh->th))
1543   {
1544     /* queue empty, no point in asking for transmission */
1545     GNUNET_CLIENT_notify_transmit_ready_cancel (mesh->th);
1546     mesh->th = NULL;
1547   }
1548 }
1549
1550
1551 void
1552 GNUNET_MESH_receive_done (struct GNUNET_MESH_Channel *channel)
1553 {
1554   send_ack (channel);
1555 }
1556
1557
1558 /**
1559  * Request information about the running mesh peer.
1560  * The callback will be called for every channel known to the service,
1561  * listing all active peers that blong to the channel.
1562  *
1563  * If called again on the same handle, it will overwrite the previous
1564  * callback and cls. To retrieve the cls, monitor_cancel must be
1565  * called first.
1566  *
1567  * WARNING: unstable API, likely to change in the future!
1568  *
1569  * @param h Handle to the mesh peer.
1570  * @param callback Function to call with the requested data.
1571  * @param callback_cls Closure for @c callback.
1572  */
1573 void
1574 GNUNET_MESH_get_channels (struct GNUNET_MESH_Handle *h,
1575                          GNUNET_MESH_ChannelsCB callback,
1576                          void *callback_cls)
1577 {
1578   struct GNUNET_MessageHeader msg;
1579
1580   msg.size = htons (sizeof (msg));
1581   msg.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNELS);
1582   send_packet (h, &msg, NULL);
1583   h->channels_cb = callback;
1584   h->channels_cls = callback_cls;
1585 }
1586
1587
1588 /**
1589  * Cancel a monitor request. The monitor callback will not be called.
1590  *
1591  * @param h Mesh handle.
1592  *
1593  * @return Closure given to GNUNET_MESH_monitor, if any.
1594  */
1595 void *
1596 GNUNET_MESH_get_channels_cancel (struct GNUNET_MESH_Handle *h)
1597 {
1598   void *cls;
1599
1600   cls = h->channels_cls;
1601   h->channels_cb = NULL;
1602   h->channels_cls = NULL;
1603   return cls;
1604 }
1605
1606
1607 /**
1608  * Request information about a specific channel of the running mesh peer.
1609  *
1610  * WARNING: unstable API, likely to change in the future!
1611  * FIXME Add destination option.
1612  *
1613  * @param h Handle to the mesh peer.
1614  * @param initiator ID of the owner of the channel.
1615  * @param channel_number Channel number.
1616  * @param callback Function to call with the requested data.
1617  * @param callback_cls Closure for @c callback.
1618  */
1619 void
1620 GNUNET_MESH_show_channel (struct GNUNET_MESH_Handle *h,
1621                          struct GNUNET_PeerIdentity *initiator,
1622                          unsigned int channel_number,
1623                          GNUNET_MESH_ChannelCB callback,
1624                          void *callback_cls)
1625 {
1626   struct GNUNET_MESH_LocalMonitor msg;
1627
1628   msg.header.size = htons (sizeof (msg));
1629   msg.header.type = htons (GNUNET_MESSAGE_TYPE_MESH_LOCAL_INFO_CHANNEL);
1630   msg.owner = *initiator;
1631   msg.channel_id = htonl (channel_number);
1632   msg.reserved = 0;
1633   send_packet (h, &msg.header, NULL);
1634   h->channel_cb = callback;
1635   h->channel_cls = callback_cls;
1636 }
1637
1638
1639 /**
1640  * Function called to notify a client about the connection
1641  * begin ready to queue more data.  "buf" will be
1642  * NULL and "size" zero if the connection was closed for
1643  * writing in the meantime.
1644  *
1645  * @param cls closure
1646  * @param size number of bytes available in buf
1647  * @param buf where the callee should write the message
1648  * @return number of bytes written to buf
1649  */
1650 static size_t
1651 mesh_mq_ntr (void *cls, size_t size,
1652              void *buf)
1653 {
1654   struct GNUNET_MQ_Handle *mq = cls;
1655   struct MeshMQState *state = GNUNET_MQ_impl_state (mq);
1656   const struct GNUNET_MessageHeader *msg = GNUNET_MQ_impl_current (mq);
1657   uint16_t msize;
1658
1659   state->th = NULL;
1660   if (NULL == buf)
1661   {
1662     GNUNET_MQ_inject_error (mq, GNUNET_MQ_ERROR_WRITE);
1663     return 0;
1664   }
1665   msize = ntohs (msg->size);
1666   GNUNET_assert (msize <= size);
1667   memcpy (buf, msg, msize);
1668   GNUNET_MQ_impl_send_continue (mq);
1669   return msize;
1670 }
1671
1672
1673 /**
1674  * Signature of functions implementing the
1675  * sending functionality of a message queue.
1676  *
1677  * @param mq the message queue
1678  * @param msg the message to send
1679  * @param impl_state state of the implementation
1680  */
1681 static void
1682 mesh_mq_send_impl (struct GNUNET_MQ_Handle *mq,
1683                    const struct GNUNET_MessageHeader *msg, void *impl_state)
1684 {
1685   struct MeshMQState *state = impl_state;
1686
1687   GNUNET_assert (NULL == state->th);
1688   state->th =
1689       GNUNET_MESH_notify_transmit_ready (state->channel,
1690                                          /* FIXME: add option for corking */
1691                                          GNUNET_NO,
1692                                          GNUNET_TIME_UNIT_FOREVER_REL,
1693                                          ntohs (msg->size),
1694                                          mesh_mq_ntr, mq);
1695
1696 }
1697
1698
1699 /**
1700  * Signature of functions implementing the
1701  * destruction of a message queue.
1702  * Implementations must not free 'mq', but should
1703  * take care of 'impl_state'.
1704  *
1705  * @param mq the message queue to destroy
1706  * @param impl_state state of the implementation
1707  */
1708 static void
1709 mesh_mq_destroy_impl (struct GNUNET_MQ_Handle *mq, void *impl_state)
1710 {
1711   struct MeshMQState *state = impl_state;
1712
1713   if (NULL != state->th)
1714     GNUNET_MESH_notify_transmit_ready_cancel (state->th);
1715
1716   GNUNET_free (state);
1717 }
1718
1719
1720 /**
1721  * Create a message queue for a mesh channel.
1722  * The message queue can only be used to transmit messages,
1723  * not to receive them.
1724  *
1725  * @param channel the channel to create the message qeue for
1726  * @return a message queue to messages over the channel
1727  */
1728 struct GNUNET_MQ_Handle *
1729 GNUNET_MESH_mq_create (struct GNUNET_MESH_Channel *channel)
1730 {
1731   struct GNUNET_MQ_Handle *mq;
1732   struct MeshMQState *state;
1733
1734   state = GNUNET_new (struct MeshMQState);
1735   state->channel = channel;
1736
1737   mq = GNUNET_MQ_queue_for_callbacks (mesh_mq_send_impl,
1738                                       mesh_mq_destroy_impl,
1739                                       NULL, /* FIXME: cancel impl. */
1740                                       state,
1741                                       NULL, /* no msg handlers */
1742                                       NULL, /* no err handlers */
1743                                       NULL); /* no handler cls */
1744   return mq;
1745 }
1746