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