src/cadet/cadet_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2011, 2017 GNUnet e.V.
   4
   5      GNUnet is free software: you can redistribute it and/or modify it
   6      under the terms of the GNU Affero General Public License as published
   7      by the Free Software Foundation, either version 3 of the License,
   8      or (at your option) any later version.
   9
  10      GNUnet is distributed in the hope that it will be useful, but
  11      WITHOUT ANY WARRANTY; without even the implied warranty of
  12      MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  13      Affero General Public License for more details.
  14
  15      You should have received a copy of the GNU Affero General Public License
  16      along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17
  18      SPDX-License-Identifier: AGPL3.0-or-later
  19 */
  20 /**
  21  * @file cadet/cadet_api.c
  22  * @brief cadet api: client implementation of cadet service
  23  * @author Bartlomiej Polot
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_constants.h"
  29 #include "gnunet_cadet_service.h"
  30 #include "cadet.h"
  31 #include "cadet_protocol.h"
  32
  33 #define LOG(kind,...) GNUNET_log_from (kind, "cadet-api",__VA_ARGS__)
  34
  35 /**
  36  * Opaque handle to the service.
  37  */
  38 struct GNUNET_CADET_Handle
  39 {
  40   /**
  41    * Message queue.
  42    */
  43   struct GNUNET_MQ_Handle *mq;
  44
  45   /**
  46    * Ports open.
  47    */
  48   struct GNUNET_CONTAINER_MultiHashMap *ports;
  49
  50   /**
  51    * Channels open.
  52    */
  53   struct GNUNET_CONTAINER_MultiHashMap32 *channels;
  54
  55   /**
  56    * child of the next channel to create (to avoid reusing IDs often)
  57    */
  58   struct GNUNET_CADET_ClientChannelNumber next_ccn;
  59
  60   /**
  61    * Configuration given by the client, in case of reconnection
  62    */
  63   const struct GNUNET_CONFIGURATION_Handle *cfg;
  64
  65   /**
  66    * Task for trying to reconnect.
  67    */
  68   struct GNUNET_SCHEDULER_Task *reconnect_task;
  69
  70   /**
  71    * Time to the next reconnect in case one reconnect fails
  72    */
  73   struct GNUNET_TIME_Relative reconnect_time;
  74
  75 };
  76
  77
  78 /**
  79  * Opaque handle to a channel.
  80  */
  81 struct GNUNET_CADET_Channel
  82 {
  83
  84   /**
  85    * Other end of the channel.
  86    */
  87   struct GNUNET_PeerIdentity peer;
  88
  89   /**
  90    * Handle to the cadet this channel belongs to
  91    */
  92   struct GNUNET_CADET_Handle *cadet;
  93
  94   /**
  95    * Channel's port, if incoming.
  96    */
  97   struct GNUNET_CADET_Port *incoming_port;
  98
  99   /**
 100    * Any data the caller wants to put in here, used for the
 101    * various callbacks (@e disconnects, @e window_changes, handlers).
 102    */
 103   void *ctx;
 104
 105   /**
 106    * Message Queue for the channel (which we are implementing).
 107    */
 108   struct GNUNET_MQ_Handle *mq;
 109
 110   /**
 111    * Task to allow mq to send more traffic.
 112    */
 113   struct GNUNET_SCHEDULER_Task *mq_cont;
 114
 115   /**
 116    * Pending envelope with a message to be transmitted to the
 117    * service as soon as we are allowed to.  Should only be
 118    * non-NULL if @e allow_send is 0.
 119    */
 120   struct GNUNET_MQ_Envelope *pending_env;
 121
 122   /**
 123    * Window change handler.
 124    */
 125   GNUNET_CADET_WindowSizeEventHandler window_changes;
 126
 127   /**
 128    * Disconnect handler.
 129    */
 130   GNUNET_CADET_DisconnectEventHandler disconnects;
 131
 132   /**
 133    * Local ID of the channel, #GNUNET_CADET_LOCAL_CHANNEL_ID_CLI bit is set if outbound.
 134    */
 135   struct GNUNET_CADET_ClientChannelNumber ccn;
 136
 137   /**
 138    * Channel options: reliability, etc.
 139    */
 140   enum GNUNET_CADET_ChannelOption options;
 141
 142   /**
 143    * How many messages are we allowed to send to the service right now?
 144    */
 145   unsigned int allow_send;
 146
 147 };
 148
 149
 150 /**
 151  * Opaque handle to a port.
 152  */
 153 struct GNUNET_CADET_Port
 154 {
 155
 156   /**
 157    * Port "number"
 158    */
 159   struct GNUNET_HashCode id;
 160
 161   /**
 162    * Handle to the CADET session this port belongs to.
 163    */
 164   struct GNUNET_CADET_Handle *cadet;
 165
 166   /**
 167    * Closure for @a handler.
 168    */
 169   void *cls;
 170
 171   /**
 172    * Handler for incoming channels on this port
 173    */
 174   GNUNET_CADET_ConnectEventHandler connects;
 175
 176   /**
 177    * Closure for @ref connects
 178    */
 179   void *connects_cls;
 180
 181   /**
 182    * Window size change handler.
 183    */
 184   GNUNET_CADET_WindowSizeEventHandler window_changes;
 185
 186   /**
 187    * Handler called when an incoming channel is destroyed.
 188    */
 189   GNUNET_CADET_DisconnectEventHandler disconnects;
 190
 191   /**
 192    * Payload handlers for incoming channels.
 193    */
 194   struct GNUNET_MQ_MessageHandler *handlers;
 195 };
 196
 197
 198 /**
 199  * Find the Port struct for a hash.
 200  *
 201  * @param h CADET handle.
 202  * @param hash HashCode for the port number.
 203  * @return The port handle if known, NULL otherwise.
 204  */
 205 static struct GNUNET_CADET_Port *
 206 find_port (const struct GNUNET_CADET_Handle *h,
 207            const struct GNUNET_HashCode *hash)
 208 {
 209   return GNUNET_CONTAINER_multihashmap_get (h->ports,
 210                                             hash);
 211 }
 212
 213
 214 /**
 215  * Get the channel handler for the channel specified by id from the given handle
 216  *
 217  * @param h Cadet handle
 218  * @param ccn ID of the wanted channel
 219  * @return handle to the required channel or NULL if not found
 220  */
 221 static struct GNUNET_CADET_Channel *
 222 find_channel (struct GNUNET_CADET_Handle *h,
 223               struct GNUNET_CADET_ClientChannelNumber ccn)
 224 {
 225   return GNUNET_CONTAINER_multihashmap32_get (h->channels,
 226                                               ntohl (ccn.channel_of_client));
 227 }
 228
 229
 230 /**
 231  * Create a new channel and insert it in the channel list of the cadet handle
 232  *
 233  * @param h Cadet handle
 234  * @param ccnp pointer to desired ccn of the channel, NULL to assign one automatically.
 235  * @return Handle to the created channel.
 236  */
 237 static struct GNUNET_CADET_Channel *
 238 create_channel (struct GNUNET_CADET_Handle *h,
 239                 const struct GNUNET_CADET_ClientChannelNumber *ccnp)
 240 {
 241   struct GNUNET_CADET_Channel *ch;
 242   struct GNUNET_CADET_ClientChannelNumber ccn;
 243
 244   ch = GNUNET_new (struct GNUNET_CADET_Channel);
 245   ch->cadet = h;
 246   if (NULL == ccnp)
 247   {
 248     while (NULL !=
 249            find_channel (h,
 250                          h->next_ccn))
 251       h->next_ccn.channel_of_client
 252         = htonl (GNUNET_CADET_LOCAL_CHANNEL_ID_CLI | (1 + ntohl (h->next_ccn.channel_of_client)));
 253     ccn = h->next_ccn;
 254   }
 255   else
 256   {
 257     ccn = *ccnp;
 258   }
 259   ch->ccn = ccn;
 260   GNUNET_assert (GNUNET_OK ==
 261                  GNUNET_CONTAINER_multihashmap32_put (h->channels,
 262                                                       ntohl (ch->ccn.channel_of_client),
 263                                                       ch,
 264                                                       GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 265   return ch;
 266 }
 267
 268
 269 /**
 270  * Destroy the specified channel.
 271  * - Destroys all peers, calling the disconnect callback on each if needed
 272  * - Cancels all outgoing traffic for that channel, calling respective notifys
 273  * - Calls cleaner if channel was inbound
 274  * - Frees all memory used
 275  *
 276  * @param ch Pointer to the channel.
 277  * @param call_cleaner Whether to call the cleaner handler.
 278  */
 279 static void
 280 destroy_channel (struct GNUNET_CADET_Channel *ch)
 281 {
 282   struct GNUNET_CADET_Handle *h = ch->cadet;
 283
 284   LOG (GNUNET_ERROR_TYPE_DEBUG,
 285        "Destroying channel %X of %p\n",
 286        htonl (ch->ccn.channel_of_client),
 287        h);
 288   GNUNET_assert (GNUNET_YES ==
 289                  GNUNET_CONTAINER_multihashmap32_remove (h->channels,
 290                                                          ntohl (ch->ccn.channel_of_client),
 291                                                          ch));
 292   if (NULL != ch->mq_cont)
 293   {
 294     GNUNET_SCHEDULER_cancel (ch->mq_cont);
 295     ch->mq_cont = NULL;
 296   }
 297   /* signal channel destruction */
 298   if (NULL != ch->disconnects)
 299     ch->disconnects (ch->ctx,
 300                      ch);
 301   if (NULL != ch->pending_env)
 302     GNUNET_MQ_discard (ch->pending_env);
 303   GNUNET_MQ_destroy (ch->mq);
 304   GNUNET_free (ch);
 305 }
 306
 307
 308 /**
 309  * Reconnect to the service, retransmit all infomation to try to restore the
 310  * original state.
 311  *
 312  * @param h handle to the cadet
 313  */
 314 static void
 315 reconnect (struct GNUNET_CADET_Handle *h);
 316
 317
 318 /**
 319  * Function called during #reconnect_cbk() to (re)open
 320  * all ports that are still open.
 321  *
 322  * @param cls the `struct GNUNET_CADET_Handle`
 323  * @param id port ID
 324  * @param value a `struct GNUNET_CADET_Channel` to open
 325  * @return #GNUNET_OK (continue to iterate)
 326  */
 327 static int
 328 open_port_cb (void *cls,
 329               const struct GNUNET_HashCode *id,
 330               void *value)
 331 {
 332   struct GNUNET_CADET_Handle *h = cls;
 333   struct GNUNET_CADET_Port *port = value;
 334   struct GNUNET_CADET_PortMessage *msg;
 335   struct GNUNET_MQ_Envelope *env;
 336
 337   (void) id;
 338   env = GNUNET_MQ_msg (msg,
 339                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_PORT_OPEN);
 340   msg->port = port->id;
 341   GNUNET_MQ_send (h->mq,
 342                   env);
 343   return GNUNET_OK;
 344 }
 345
 346
 347 /**
 348  * Reconnect callback: tries to reconnect again after a failer previous
 349  * reconnecttion
 350  *
 351  * @param cls closure (cadet handle)
 352  */
 353 static void
 354 reconnect_cbk (void *cls)
 355 {
 356   struct GNUNET_CADET_Handle *h = cls;
 357
 358   h->reconnect_task = NULL;
 359   h->reconnect_time
 360     = GNUNET_TIME_STD_BACKOFF (h->reconnect_time);
 361   reconnect (h);
 362   GNUNET_CONTAINER_multihashmap_iterate (h->ports,
 363                                          &open_port_cb,
 364                                          h);
 365 }
 366
 367
 368 /**
 369  * Notify the application about a change in the window size (if needed).
 370  *
 371  * @param ch Channel to notify about.
 372  */
 373 static void
 374 notify_window_size (struct GNUNET_CADET_Channel *ch)
 375 {
 376   if (NULL != ch->window_changes)
 377     ch->window_changes (ch->ctx,
 378                         ch, /* FIXME: remove 'ch'? */
 379                         ch->allow_send);
 380 }
 381
 382
 383 /**
 384  * Transmit the next message from our queue.
 385  *
 386  * @param cls Closure (channel whose mq to activate).
 387  */
 388 static void
 389 cadet_mq_send_now (void *cls)
 390 {
 391   struct GNUNET_CADET_Channel *ch = cls;
 392   struct GNUNET_MQ_Envelope *env = ch->pending_env;
 393
 394   ch->mq_cont = NULL;
 395   if (0 == ch->allow_send)
 396   {
 397     /* how did we get here? */
 398     GNUNET_break (0);
 399     return;
 400   }
 401   if (NULL == env)
 402   {
 403     /* how did we get here? */
 404     GNUNET_break (0);
 405     return;
 406   }
 407   ch->allow_send--;
 408   ch->pending_env = NULL;
 409   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 410               "Sending message on channel %s to CADET, new window size is %u\n",
 411               GNUNET_i2s (&ch->peer),
 412               ch->allow_send);
 413   GNUNET_MQ_send (ch->cadet->mq,
 414                   env);
 415   GNUNET_MQ_impl_send_continue (ch->mq);
 416 }
 417
 418
 419 /**
 420  * Implement sending functionality of a message queue for
 421  * us sending messages to a peer.
 422  *
 423  * Encapsulates the payload message in a #GNUNET_CADET_LocalData message
 424  * in order to label the message with the channel ID and send the
 425  * encapsulated message to the service.
 426  *
 427  * @param mq the message queue
 428  * @param msg the message to send
 429  * @param impl_state state of the implementation
 430  */
 431 static void
 432 cadet_mq_send_impl (struct GNUNET_MQ_Handle *mq,
 433                     const struct GNUNET_MessageHeader *msg,
 434                     void *impl_state)
 435 {
 436   struct GNUNET_CADET_Channel *ch = impl_state;
 437   struct GNUNET_CADET_Handle *h = ch->cadet;
 438   uint16_t msize;
 439   struct GNUNET_MQ_Envelope *env;
 440   struct GNUNET_CADET_LocalData *cadet_msg = NULL;
 441
 442   if (NULL == h->mq)
 443   {
 444     /* We're currently reconnecting, pretend this worked */
 445     GNUNET_MQ_impl_send_continue (mq);
 446     return;
 447   }
 448
 449   /* check message size for sanity */
 450   msize = ntohs (msg->size);
 451   if (msize > GNUNET_CONSTANTS_MAX_CADET_MESSAGE_SIZE)
 452   {
 453     GNUNET_break (0);
 454     GNUNET_MQ_impl_send_continue (mq);
 455     return;
 456   }
 457   env = GNUNET_MQ_msg_nested_mh (cadet_msg,
 458                                  GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA,
 459                                  msg);
 460   cadet_msg->ccn = ch->ccn;
 461   GNUNET_assert (NULL == ch->pending_env);
 462   ch->pending_env = env;
 463   if (0 < ch->allow_send)
 464     ch->mq_cont
 465       = GNUNET_SCHEDULER_add_now (&cadet_mq_send_now,
 466                                   ch);
 467 }
 468
 469
 470 /**
 471  * Handle destruction of a message queue.  Implementations must not
 472  * free @a mq, but should take care of @a impl_state.
 473  *
 474  * @param mq the message queue to destroy
 475  * @param impl_state state of the implementation
 476  */
 477 static void
 478 cadet_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
 479                        void *impl_state)
 480 {
 481   struct GNUNET_CADET_Channel *ch = impl_state;
 482
 483   GNUNET_assert (mq == ch->mq);
 484   ch->mq = NULL;
 485 }
 486
 487
 488 /**
 489  * We had an error processing a message we forwarded from a peer to
 490  * the CADET service.  We should just complain about it but otherwise
 491  * continue processing.
 492  *
 493  * @param cls closure with our `struct GNUNET_CADET_Channel`
 494  * @param error error code
 495  */
 496 static void
 497 cadet_mq_error_handler (void *cls,
 498                         enum GNUNET_MQ_Error error)
 499 {
 500   struct GNUNET_CADET_Channel *ch = cls;
 501
 502   if (GNUNET_MQ_ERROR_NO_MATCH == error)
 503   {
 504     /* Got a message we did not understand, still try to continue! */
 505     GNUNET_break_op (0);
 506     GNUNET_CADET_receive_done (ch);
 507   }
 508   else
 509   {
 510     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 511                 "MQ error in communication with CADET: %d\n",
 512                 error);
 513     if (NULL != ch->disconnects)
 514       ch->disconnects (ch->ctx,
 515                        ch);
 516     GNUNET_CADET_channel_destroy (ch);
 517   }
 518 }
 519
 520
 521 /**
 522  * Implementation function that cancels the currently sent message.
 523  * Should basically undo whatever #mq_send_impl() did.
 524  *
 525  * @param mq message queue
 526  * @param impl_state state specific to the implementation
 527  */
 528 static void
 529 cadet_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
 530                      void *impl_state)
 531 {
 532   struct GNUNET_CADET_Channel *ch = impl_state;
 533
 534   (void) mq;
 535   GNUNET_assert (NULL != ch->pending_env);
 536   GNUNET_MQ_discard (ch->pending_env);
 537   ch->pending_env = NULL;
 538   if (NULL != ch->mq_cont)
 539   {
 540     GNUNET_SCHEDULER_cancel (ch->mq_cont);
 541     ch->mq_cont = NULL;
 542   }
 543 }
 544
 545
 546 /**
 547  * Process the new channel notification and add it to the channels in the handle
 548  *
 549  * @param h     The cadet handle
 550  * @param msg   A message with the details of the new incoming channel
 551  */
 552 static void
 553 handle_channel_created (void *cls,
 554                         const struct GNUNET_CADET_LocalChannelCreateMessage *msg)
 555 {
 556   struct GNUNET_CADET_Handle *h = cls;
 557   struct GNUNET_CADET_Channel *ch;
 558   struct GNUNET_CADET_Port *port;
 559   const struct GNUNET_HashCode *port_number;
 560   struct GNUNET_CADET_ClientChannelNumber ccn;
 561
 562   ccn = msg->ccn;
 563   port_number = &msg->port;
 564   if (ntohl (ccn.channel_of_client) >= GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
 565   {
 566     GNUNET_break (0);
 567     return;
 568   }
 569   port = find_port (h,
 570                     port_number);
 571   if (NULL == port)
 572   {
 573     /* We could have closed the port but the service didn't know about it yet
 574      * This is not an error.
 575      */
 576     struct GNUNET_CADET_LocalChannelDestroyMessage *d_msg;
 577     struct GNUNET_MQ_Envelope *env;
 578
 579     LOG (GNUNET_ERROR_TYPE_DEBUG,
 580          "No handler for incoming channel %X (on port %s, recently closed?)\n",
 581          ntohl (ccn.channel_of_client),
 582          GNUNET_h2s (port_number));
 583     env = GNUNET_MQ_msg (d_msg,
 584                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY);
 585     d_msg->ccn = msg->ccn;
 586     GNUNET_MQ_send (h->mq,
 587                     env);
 588     return;
 589   }
 590
 591   ch = create_channel (h,
 592                        &ccn);
 593   ch->peer = msg->peer;
 594   ch->incoming_port = port;
 595   ch->options = ntohl (msg->opt);
 596   LOG (GNUNET_ERROR_TYPE_DEBUG,
 597        "Creating incoming channel %X [%s] %p\n",
 598        ntohl (ccn.channel_of_client),
 599        GNUNET_h2s (port_number),
 600        ch);
 601
 602   GNUNET_assert (NULL != port->connects);
 603   ch->window_changes = port->window_changes;
 604   ch->disconnects = port->disconnects;
 605   ch->mq = GNUNET_MQ_queue_for_callbacks (&cadet_mq_send_impl,
 606                                           &cadet_mq_destroy_impl,
 607                                           &cadet_mq_cancel_impl,
 608                                           ch,
 609                                           port->handlers,
 610                                           &cadet_mq_error_handler,
 611                                           ch);
 612   ch->ctx = port->connects (port->cls,
 613                             ch,
 614                             &msg->peer);
 615   GNUNET_MQ_set_handlers_closure (ch->mq,
 616                                   ch->ctx);
 617 }
 618
 619
 620 /**
 621  * Process the channel destroy notification and free associated resources
 622  *
 623  * @param h     The cadet handle
 624  * @param msg   A message with the details of the channel being destroyed
 625  */
 626 static void
 627 handle_channel_destroy (void *cls,
 628                         const struct GNUNET_CADET_LocalChannelDestroyMessage *msg)
 629 {
 630   struct GNUNET_CADET_Handle *h = cls;
 631   struct GNUNET_CADET_Channel *ch;
 632
 633   ch = find_channel (h,
 634                      msg->ccn);
 635   if (NULL == ch)
 636   {
 637     LOG (GNUNET_ERROR_TYPE_DEBUG,
 638          "Received channel destroy for unknown channel %X from CADET service (recently close?)\n",
 639          ntohl (msg->ccn.channel_of_client));
 640     return;
 641   }
 642   LOG (GNUNET_ERROR_TYPE_DEBUG,
 643        "Received channel destroy for channel %X from CADET service\n",
 644        ntohl (msg->ccn.channel_of_client));
 645   destroy_channel (ch);
 646 }
 647
 648
 649 /**
 650  * Check that message received from CADET service is well-formed.
 651  *
 652  * @param cls the `struct GNUNET_CADET_Handle`
 653  * @param message the message we got
 654  * @return #GNUNET_OK if the message is well-formed,
 655  *         #GNUNET_SYSERR otherwise
 656  */
 657 static int
 658 check_local_data (void *cls,
 659                   const struct GNUNET_CADET_LocalData *message)
 660 {
 661   uint16_t size;
 662
 663   (void) cls;
 664   size = ntohs (message->header.size);
 665   if (sizeof (*message) + sizeof (struct GNUNET_MessageHeader) > size)
 666   {
 667     GNUNET_break (0);
 668     return GNUNET_SYSERR;
 669   }
 670   return GNUNET_OK;
 671 }
 672
 673
 674 /**
 675  * Process the incoming data packets, call appropriate handlers.
 676  *
 677  * @param h       The cadet handle
 678  * @param message A message encapsulating the data
 679  */
 680 static void
 681 handle_local_data (void *cls,
 682                    const struct GNUNET_CADET_LocalData *message)
 683 {
 684   struct GNUNET_CADET_Handle *h = cls;
 685   const struct GNUNET_MessageHeader *payload;
 686   struct GNUNET_CADET_Channel *ch;
 687   uint16_t type;
 688   int fwd;
 689
 690   ch = find_channel (h,
 691                      message->ccn);
 692   if (NULL == ch)
 693   {
 694     LOG (GNUNET_ERROR_TYPE_DEBUG,
 695          "Unknown channel %X for incoming data (recently closed?)\n",
 696          ntohl (message->ccn.channel_of_client));
 697     return;
 698   }
 699
 700   payload = (const struct GNUNET_MessageHeader *) &message[1];
 701   type = ntohs (payload->type);
 702   fwd = ntohl (ch->ccn.channel_of_client) <= GNUNET_CADET_LOCAL_CHANNEL_ID_CLI;
 703   LOG (GNUNET_ERROR_TYPE_DEBUG,
 704        "Got a %s data on channel %s [%X] of type %u\n",
 705        fwd ? "FWD" : "BWD",
 706        GNUNET_i2s (&ch->peer),
 707        ntohl (message->ccn.channel_of_client),
 708        type);
 709   GNUNET_MQ_inject_message (ch->mq,
 710                             payload);
 711 }
 712
 713
 714 /**
 715  * Process a local ACK message, enabling the client to send
 716  * more data to the service.
 717  *
 718  * @param h Cadet handle.
 719  * @param message Message itself.
 720  */
 721 static void
 722 handle_local_ack (void *cls,
 723                   const struct GNUNET_CADET_LocalAck *message)
 724 {
 725   struct GNUNET_CADET_Handle *h = cls;
 726   struct GNUNET_CADET_Channel *ch;
 727
 728   ch = find_channel (h,
 729                      message->ccn);
 730   if (NULL == ch)
 731   {
 732     LOG (GNUNET_ERROR_TYPE_DEBUG,
 733          "ACK on unknown channel %X\n",
 734          ntohl (message->ccn.channel_of_client));
 735     return;
 736   }
 737   ch->allow_send++;
 738   LOG (GNUNET_ERROR_TYPE_DEBUG,
 739        "Got an ACK on mq channel %X (peer %s); new window size is %u!\n",
 740        ntohl (ch->ccn.channel_of_client),
 741        GNUNET_i2s (&ch->peer),
 742        ch->allow_send);
 743   if (NULL == ch->pending_env)
 744   {
 745     LOG (GNUNET_ERROR_TYPE_DEBUG,
 746          "Got an ACK on mq channel %X, allow send now %u!\n",
 747          ntohl (ch->ccn.channel_of_client),
 748          ch->allow_send);
 749     notify_window_size (ch);
 750     return;
 751   }
 752   if (NULL != ch->mq_cont)
 753     return; /* already working on it! */
 754   ch->mq_cont
 755     = GNUNET_SCHEDULER_add_now (&cadet_mq_send_now,
 756                                 ch);
 757 }
 758
 759
 760 /**
 761  * Function called during #GNUNET_CADET_disconnect() to destroy
 762  * all channels that are still open.
 763  *
 764  * @param cls the `struct GNUNET_CADET_Handle`
 765  * @param cid chanenl ID
 766  * @param value a `struct GNUNET_CADET_Channel` to destroy
 767  * @return #GNUNET_OK (continue to iterate)
 768  */
 769 static int
 770 destroy_channel_cb (void *cls,
 771                     uint32_t cid,
 772                     void *value)
 773 {
 774   /* struct GNUNET_CADET_Handle *handle = cls; */
 775   struct GNUNET_CADET_Channel *ch = value;
 776
 777   (void) cls;
 778   (void) cid;
 779   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 780               "Destroying channel due to GNUNET_CADET_disconnect()\n");
 781   destroy_channel (ch);
 782   return GNUNET_OK;
 783 }
 784
 785
 786 /**
 787  * Generic error handler, called with the appropriate error code and
 788  * the same closure specified at the creation of the message queue.
 789  * Not every message queue implementation supports an error handler.
 790  *
 791  * @param cls closure, a `struct GNUNET_CORE_Handle *`
 792  * @param error error code
 793  */
 794 static void
 795 handle_mq_error (void *cls,
 796                  enum GNUNET_MQ_Error error)
 797 {
 798   struct GNUNET_CADET_Handle *h = cls;
 799
 800   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 801               "MQ ERROR: %u\n",
 802               error);
 803   GNUNET_CONTAINER_multihashmap32_iterate (h->channels,
 804                                            &destroy_channel_cb,
 805                                            h);
 806   GNUNET_MQ_destroy (h->mq);
 807   h->mq = NULL;
 808   GNUNET_assert (NULL == h->reconnect_task);
 809   h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 810                                                     &reconnect_cbk,
 811                                                     h);
 812 }
 813
 814
 815 /**
 816  * Reconnect to the service, retransmit all infomation to try to restore the
 817  * original state.
 818  *
 819  * @param h handle to the cadet
 820  */
 821 static void
 822 reconnect (struct GNUNET_CADET_Handle *h)
 823 {
 824   struct GNUNET_MQ_MessageHandler handlers[] = {
 825     GNUNET_MQ_hd_fixed_size (channel_created,
 826                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE,
 827                              struct GNUNET_CADET_LocalChannelCreateMessage,
 828                              h),
 829     GNUNET_MQ_hd_fixed_size (channel_destroy,
 830                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY,
 831                              struct GNUNET_CADET_LocalChannelDestroyMessage,
 832                              h),
 833     GNUNET_MQ_hd_var_size (local_data,
 834                            GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA,
 835                            struct GNUNET_CADET_LocalData,
 836                            h),
 837     GNUNET_MQ_hd_fixed_size (local_ack,
 838                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK,
 839                              struct GNUNET_CADET_LocalAck,
 840                              h),
 841     GNUNET_MQ_handler_end ()
 842   };
 843
 844   GNUNET_assert (NULL == h->mq);
 845   h->mq = GNUNET_CLIENT_connect (h->cfg,
 846                                  "cadet",
 847                                  handlers,
 848                                  &handle_mq_error,
 849                                  h);
 850 }
 851
 852
 853 /**
 854  * Function called during #GNUNET_CADET_disconnect() to destroy
 855  * all ports that are still open.
 856  *
 857  * @param cls the `struct GNUNET_CADET_Handle`
 858  * @param id port ID
 859  * @param value a `struct GNUNET_CADET_Channel` to destroy
 860  * @return #GNUNET_OK (continue to iterate)
 861  */
 862 static int
 863 destroy_port_cb (void *cls,
 864                  const struct GNUNET_HashCode *id,
 865                  void *value)
 866 {
 867   /* struct GNUNET_CADET_Handle *handle = cls; */
 868   struct GNUNET_CADET_Port *port = value;
 869
 870   (void) cls;
 871   (void) id;
 872   /* This is a warning, the app should have cleanly closed all open ports */
 873   GNUNET_break (0);
 874   GNUNET_CADET_close_port (port);
 875   return GNUNET_OK;
 876 }
 877
 878
 879 /**
 880  * Disconnect from the cadet service. All channels will be destroyed. All channel
 881  * disconnect callbacks will be called on any still connected peers, notifying
 882  * about their disconnection. The registered inbound channel cleaner will be
 883  * called should any inbound channels still exist.
 884  *
 885  * @param handle connection to cadet to disconnect
 886  */
 887 void
 888 GNUNET_CADET_disconnect (struct GNUNET_CADET_Handle *handle)
 889 {
 890   GNUNET_CONTAINER_multihashmap_iterate (handle->ports,
 891                                          &destroy_port_cb,
 892                                          handle);
 893   GNUNET_CONTAINER_multihashmap_destroy (handle->ports);
 894   handle->ports = NULL;
 895   GNUNET_CONTAINER_multihashmap32_iterate (handle->channels,
 896                                            &destroy_channel_cb,
 897                                            handle);
 898   GNUNET_CONTAINER_multihashmap32_destroy (handle->channels);
 899   handle->channels = NULL;
 900   if (NULL != handle->mq)
 901   {
 902     GNUNET_MQ_destroy (handle->mq);
 903     handle->mq = NULL;
 904   }
 905   if (NULL != handle->reconnect_task)
 906   {
 907     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 908     handle->reconnect_task = NULL;
 909   }
 910   GNUNET_free (handle);
 911 }
 912
 913
 914 /**
 915  * Close a port opened with @a GNUNET_CADET_open_port().
 916  * The @a new_channel callback will no longer be called.
 917  *
 918  * @param p Port handle.
 919  */
 920 void
 921 GNUNET_CADET_close_port (struct GNUNET_CADET_Port *p)
 922 {
 923   GNUNET_assert (GNUNET_YES ==
 924                  GNUNET_CONTAINER_multihashmap_remove (p->cadet->ports,
 925                                                        &p->id,
 926                                                        p));
 927   if (NULL != p->cadet->mq)
 928   {
 929     struct GNUNET_CADET_PortMessage *msg;
 930     struct GNUNET_MQ_Envelope *env;
 931
 932     env = GNUNET_MQ_msg (msg,
 933                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_PORT_CLOSE);
 934     msg->port = p->id;
 935     GNUNET_MQ_send (p->cadet->mq,
 936                     env);
 937   }
 938   GNUNET_free_non_null (p->handlers);
 939   GNUNET_free (p);
 940 }
 941
 942
 943 /**
 944  * Destroy an existing channel.
 945  *
 946  * The existing end callback for the channel will NOT be called.
 947  * Any pending outgoing messages will be sent but no incoming messages will be
 948  * accepted and no data callbacks will be called.
 949  *
 950  * @param channel Channel handle, becomes invalid after this call.
 951  */
 952 void
 953 GNUNET_CADET_channel_destroy (struct GNUNET_CADET_Channel *channel)
 954 {
 955   struct GNUNET_CADET_Handle *h = channel->cadet;
 956   struct GNUNET_CADET_LocalChannelDestroyMessage *msg;
 957   struct GNUNET_MQ_Envelope *env;
 958
 959   if (NULL != h->mq)
 960   {
 961     env = GNUNET_MQ_msg (msg,
 962                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY);
 963     msg->ccn = channel->ccn;
 964     GNUNET_MQ_send (h->mq,
 965                     env);
 966   }
 967   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 968               "Destroying channel due to GNUNET_CADET_channel_destroy()\n");
 969   channel->disconnects = NULL;
 970   destroy_channel (channel);
 971 }
 972
 973
 974 /**
 975  * Get information about a channel.
 976  *
 977  * @param channel Channel handle.
 978  * @param option Query (GNUNET_CADET_OPTION_*).
 979  * @param ... dependant on option, currently not used
 980  *
 981  * @return Union with an answer to the query.
 982  */
 983 const union GNUNET_CADET_ChannelInfo *
 984 GNUNET_CADET_channel_get_info (struct GNUNET_CADET_Channel *channel,
 985                                enum GNUNET_CADET_ChannelOption option,
 986                                ...)
 987 {
 988   static int bool_flag;
 989
 990   switch (option)
 991   {
 992     case GNUNET_CADET_OPTION_NOBUFFER:
 993     case GNUNET_CADET_OPTION_RELIABLE:
 994     case GNUNET_CADET_OPTION_OUT_OF_ORDER:
 995       if (0 != (option & channel->options))
 996         bool_flag = GNUNET_YES;
 997       else
 998         bool_flag = GNUNET_NO;
 999       return (const union GNUNET_CADET_ChannelInfo *) &bool_flag;
1000       break;
1001     case GNUNET_CADET_OPTION_PEER:
1002       return (const union GNUNET_CADET_ChannelInfo *) &channel->peer;
1003       break;
1004     default:
1005       GNUNET_break (0);
1006       return NULL;
1007   }
1008 }
1009
1010
1011 /**
1012  * Send an ack on the channel to confirm the processing of a message.
1013  *
1014  * @param ch Channel on which to send the ACK.
1015  */
1016 void
1017 GNUNET_CADET_receive_done (struct GNUNET_CADET_Channel *channel)
1018 {
1019   struct GNUNET_CADET_LocalAck *msg;
1020   struct GNUNET_MQ_Envelope *env;
1021
1022   env = GNUNET_MQ_msg (msg,
1023                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
1024   LOG (GNUNET_ERROR_TYPE_DEBUG,
1025        "Sending ACK on channel %X\n",
1026        ntohl (channel->ccn.channel_of_client));
1027   msg->ccn = channel->ccn;
1028   GNUNET_MQ_send (channel->cadet->mq,
1029                   env);
1030 }
1031
1032
1033 /**
1034  * Connect to the MQ-based cadet service.
1035  *
1036  * @param cfg Configuration to use.
1037  *
1038  * @return Handle to the cadet service NULL on error.
1039  */
1040 struct GNUNET_CADET_Handle *
1041 GNUNET_CADET_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
1042 {
1043   struct GNUNET_CADET_Handle *h;
1044
1045   LOG (GNUNET_ERROR_TYPE_DEBUG,
1046        "GNUNET_CADET_connect()\n");
1047   h = GNUNET_new (struct GNUNET_CADET_Handle);
1048   h->cfg = cfg;
1049   h->ports = GNUNET_CONTAINER_multihashmap_create (4,
1050                                                    GNUNET_YES);
1051   h->channels = GNUNET_CONTAINER_multihashmap32_create (4);
1052   reconnect (h);
1053   if (NULL == h->mq)
1054   {
1055     GNUNET_break (0);
1056     GNUNET_CADET_disconnect (h);
1057     return NULL;
1058   }
1059   h->next_ccn.channel_of_client = htonl (GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
1060   return h;
1061 }
1062
1063
1064 /**
1065  * Open a port to receive incomming MQ-based channels.
1066  *
1067  * @param h CADET handle.
1068  * @param port Hash identifying the port.
1069  * @param connects Function called when an incoming channel is connected.
1070  * @param connects_cls Closure for the @a connects handler.
1071  * @param window_changes Function called when the transmit window size changes.
1072  * @param disconnects Function called when a channel is disconnected.
1073  * @param handlers Callbacks for messages we care about, NULL-terminated.
1074  * @return Port handle, NULL if port is in use
1075  */
1076 struct GNUNET_CADET_Port *
1077 GNUNET_CADET_open_port (struct GNUNET_CADET_Handle *h,
1078                         const struct GNUNET_HashCode *port,
1079                         GNUNET_CADET_ConnectEventHandler connects,
1080                         void * connects_cls,
1081                         GNUNET_CADET_WindowSizeEventHandler window_changes,
1082                         GNUNET_CADET_DisconnectEventHandler disconnects,
1083                         const struct GNUNET_MQ_MessageHandler *handlers)
1084 {
1085   struct GNUNET_CADET_Port *p;
1086
1087   GNUNET_assert (NULL != connects);
1088   GNUNET_assert (NULL != disconnects);
1089   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1090               "Listening to CADET port %s\n",
1091               GNUNET_h2s (port));
1092
1093   p = GNUNET_new (struct GNUNET_CADET_Port);
1094   p->cadet = h;
1095   p->id = *port;
1096   if (GNUNET_OK !=
1097       GNUNET_CONTAINER_multihashmap_put (h->ports,
1098                                          &p->id,
1099                                          p,
1100                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
1101   {
1102     GNUNET_free (p);
1103     return NULL;
1104   }
1105   p->connects = connects;
1106   p->cls = connects_cls;
1107   p->window_changes = window_changes;
1108   p->disconnects = disconnects;
1109   p->handlers = GNUNET_MQ_copy_handlers (handlers);
1110
1111   GNUNET_assert (GNUNET_OK ==
1112                  open_port_cb (h,
1113                                &p->id,
1114                                p));
1115   return p;
1116 }
1117
1118
1119 /**
1120  * Create a new channel towards a remote peer.
1121  *
1122  * If the destination port is not open by any peer or the destination peer
1123  * does not accept the channel, #GNUNET_CADET_ChannelEndHandler will be called
1124  * for this channel.
1125  *
1126  * @param h CADET handle.
1127  * @param channel_cls Closure for the channel. It's given to:
1128  *                    - The disconnect handler @a disconnects
1129  *                    - Each message type callback in @a handlers
1130  * @param destination Peer identity the channel should go to.
1131  * @param port Identification of the destination port.
1132  * @param options CadetOption flag field, with all desired option bits set to 1.
1133  * @param window_changes Function called when the transmit window size changes.
1134  * @param disconnects Function called when the channel is disconnected.
1135  * @param handlers Callbacks for messages we care about, NULL-terminated.
1136  * @return Handle to the channel.
1137  */
1138 struct GNUNET_CADET_Channel *
1139 GNUNET_CADET_channel_create (struct GNUNET_CADET_Handle *h,
1140                              void *channel_cls,
1141                              const struct GNUNET_PeerIdentity *destination,
1142                              const struct GNUNET_HashCode *port,
1143                              enum GNUNET_CADET_ChannelOption options,
1144                              GNUNET_CADET_WindowSizeEventHandler window_changes,
1145                              GNUNET_CADET_DisconnectEventHandler disconnects,
1146                              const struct GNUNET_MQ_MessageHandler *handlers)
1147 {
1148   struct GNUNET_CADET_Channel *ch;
1149   struct GNUNET_CADET_LocalChannelCreateMessage *msg;
1150   struct GNUNET_MQ_Envelope *env;
1151
1152   GNUNET_assert (NULL != disconnects);
1153   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1154               "Creating channel to peer %s at port %s\n",
1155               GNUNET_i2s (destination),
1156               GNUNET_h2s (port));
1157   ch = create_channel (h,
1158                        NULL);
1159   ch->ctx = channel_cls;
1160   ch->peer = *destination;
1161   ch->options = options;
1162   ch->window_changes = window_changes;
1163   ch->disconnects = disconnects;
1164
1165   /* Create MQ for channel */
1166   ch->mq = GNUNET_MQ_queue_for_callbacks (&cadet_mq_send_impl,
1167                                           &cadet_mq_destroy_impl,
1168                                           &cadet_mq_cancel_impl,
1169                                           ch,
1170                                           handlers,
1171                                           &cadet_mq_error_handler,
1172                                           ch);
1173   GNUNET_MQ_set_handlers_closure (ch->mq,
1174                                   channel_cls);
1175
1176   /* Request channel creation to service */
1177   env = GNUNET_MQ_msg (msg,
1178                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE);
1179   msg->ccn = ch->ccn;
1180   msg->port = *port;
1181   msg->peer = *destination;
1182   msg->opt = htonl (options);
1183   GNUNET_MQ_send (h->mq,
1184                   env);
1185   return ch;
1186 }
1187
1188
1189 /**
1190  * Obtain the message queue for a connected peer.
1191  *
1192  * @param channel The channel handle from which to get the MQ.
1193  *
1194  * @return NULL if @a channel is not yet connected.
1195  */
1196 struct GNUNET_MQ_Handle *
1197 GNUNET_CADET_get_mq (const struct GNUNET_CADET_Channel *channel)
1198 {
1199   return channel->mq;
1200 }
1201
1202 /* end of cadet_api.c */