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_break (0);
 511     if (NULL != ch->disconnects)
 512       ch->disconnects (ch->ctx,
 513                        ch);
 514     GNUNET_CADET_channel_destroy (ch);
 515   }
 516 }
 517
 518
 519 /**
 520  * Implementation function that cancels the currently sent message.
 521  * Should basically undo whatever #mq_send_impl() did.
 522  *
 523  * @param mq message queue
 524  * @param impl_state state specific to the implementation
 525  */
 526 static void
 527 cadet_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
 528                      void *impl_state)
 529 {
 530   struct GNUNET_CADET_Channel *ch = impl_state;
 531
 532   (void) mq;
 533   GNUNET_assert (NULL != ch->pending_env);
 534   GNUNET_MQ_discard (ch->pending_env);
 535   ch->pending_env = NULL;
 536   if (NULL != ch->mq_cont)
 537   {
 538     GNUNET_SCHEDULER_cancel (ch->mq_cont);
 539     ch->mq_cont = NULL;
 540   }
 541 }
 542
 543
 544 /**
 545  * Process the new channel notification and add it to the channels in the handle
 546  *
 547  * @param h     The cadet handle
 548  * @param msg   A message with the details of the new incoming channel
 549  */
 550 static void
 551 handle_channel_created (void *cls,
 552                         const struct GNUNET_CADET_LocalChannelCreateMessage *msg)
 553 {
 554   struct GNUNET_CADET_Handle *h = cls;
 555   struct GNUNET_CADET_Channel *ch;
 556   struct GNUNET_CADET_Port *port;
 557   const struct GNUNET_HashCode *port_number;
 558   struct GNUNET_CADET_ClientChannelNumber ccn;
 559
 560   ccn = msg->ccn;
 561   port_number = &msg->port;
 562   if (ntohl (ccn.channel_of_client) >= GNUNET_CADET_LOCAL_CHANNEL_ID_CLI)
 563   {
 564     GNUNET_break (0);
 565     return;
 566   }
 567   port = find_port (h,
 568                     port_number);
 569   if (NULL == port)
 570   {
 571     /* We could have closed the port but the service didn't know about it yet
 572      * This is not an error.
 573      */
 574     struct GNUNET_CADET_LocalChannelDestroyMessage *d_msg;
 575     struct GNUNET_MQ_Envelope *env;
 576
 577     LOG (GNUNET_ERROR_TYPE_DEBUG,
 578          "No handler for incoming channel %X (on port %s, recently closed?)\n",
 579          ntohl (ccn.channel_of_client),
 580          GNUNET_h2s (port_number));
 581     env = GNUNET_MQ_msg (d_msg,
 582                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY);
 583     d_msg->ccn = msg->ccn;
 584     GNUNET_MQ_send (h->mq,
 585                     env);
 586     return;
 587   }
 588
 589   ch = create_channel (h,
 590                        &ccn);
 591   ch->peer = msg->peer;
 592   ch->incoming_port = port;
 593   ch->options = ntohl (msg->opt);
 594   LOG (GNUNET_ERROR_TYPE_DEBUG,
 595        "Creating incoming channel %X [%s] %p\n",
 596        ntohl (ccn.channel_of_client),
 597        GNUNET_h2s (port_number),
 598        ch);
 599
 600   GNUNET_assert (NULL != port->connects);
 601   ch->window_changes = port->window_changes;
 602   ch->disconnects = port->disconnects;
 603   ch->mq = GNUNET_MQ_queue_for_callbacks (&cadet_mq_send_impl,
 604                                           &cadet_mq_destroy_impl,
 605                                           &cadet_mq_cancel_impl,
 606                                           ch,
 607                                           port->handlers,
 608                                           &cadet_mq_error_handler,
 609                                           ch);
 610   ch->ctx = port->connects (port->cls,
 611                             ch,
 612                             &msg->peer);
 613   GNUNET_MQ_set_handlers_closure (ch->mq,
 614                                   ch->ctx);
 615 }
 616
 617
 618 /**
 619  * Process the channel destroy notification and free associated resources
 620  *
 621  * @param h     The cadet handle
 622  * @param msg   A message with the details of the channel being destroyed
 623  */
 624 static void
 625 handle_channel_destroy (void *cls,
 626                         const struct GNUNET_CADET_LocalChannelDestroyMessage *msg)
 627 {
 628   struct GNUNET_CADET_Handle *h = cls;
 629   struct GNUNET_CADET_Channel *ch;
 630
 631   ch = find_channel (h,
 632                      msg->ccn);
 633   if (NULL == ch)
 634   {
 635     LOG (GNUNET_ERROR_TYPE_DEBUG,
 636          "Received channel destroy for unknown channel %X from CADET service (recently close?)\n",
 637          ntohl (msg->ccn.channel_of_client));
 638     return;
 639   }
 640   LOG (GNUNET_ERROR_TYPE_DEBUG,
 641        "Received channel destroy for channel %X from CADET service\n",
 642        ntohl (msg->ccn.channel_of_client));
 643   destroy_channel (ch);
 644 }
 645
 646
 647 /**
 648  * Check that message received from CADET service is well-formed.
 649  *
 650  * @param cls the `struct GNUNET_CADET_Handle`
 651  * @param message the message we got
 652  * @return #GNUNET_OK if the message is well-formed,
 653  *         #GNUNET_SYSERR otherwise
 654  */
 655 static int
 656 check_local_data (void *cls,
 657                   const struct GNUNET_CADET_LocalData *message)
 658 {
 659   uint16_t size;
 660
 661   (void) cls;
 662   size = ntohs (message->header.size);
 663   if (sizeof (*message) + sizeof (struct GNUNET_MessageHeader) > size)
 664   {
 665     GNUNET_break (0);
 666     return GNUNET_SYSERR;
 667   }
 668   return GNUNET_OK;
 669 }
 670
 671
 672 /**
 673  * Process the incoming data packets, call appropriate handlers.
 674  *
 675  * @param h       The cadet handle
 676  * @param message A message encapsulating the data
 677  */
 678 static void
 679 handle_local_data (void *cls,
 680                    const struct GNUNET_CADET_LocalData *message)
 681 {
 682   struct GNUNET_CADET_Handle *h = cls;
 683   const struct GNUNET_MessageHeader *payload;
 684   struct GNUNET_CADET_Channel *ch;
 685   uint16_t type;
 686   int fwd;
 687
 688   ch = find_channel (h,
 689                      message->ccn);
 690   if (NULL == ch)
 691   {
 692     LOG (GNUNET_ERROR_TYPE_DEBUG,
 693          "Unknown channel %X for incoming data (recently closed?)\n",
 694          ntohl (message->ccn.channel_of_client));
 695     return;
 696   }
 697
 698   payload = (const struct GNUNET_MessageHeader *) &message[1];
 699   type = ntohs (payload->type);
 700   fwd = ntohl (ch->ccn.channel_of_client) <= GNUNET_CADET_LOCAL_CHANNEL_ID_CLI;
 701   LOG (GNUNET_ERROR_TYPE_DEBUG,
 702        "Got a %s data on channel %s [%X] of type %u\n",
 703        fwd ? "FWD" : "BWD",
 704        GNUNET_i2s (&ch->peer),
 705        ntohl (message->ccn.channel_of_client),
 706        type);
 707   GNUNET_MQ_inject_message (ch->mq,
 708                             payload);
 709 }
 710
 711
 712 /**
 713  * Process a local ACK message, enabling the client to send
 714  * more data to the service.
 715  *
 716  * @param h Cadet handle.
 717  * @param message Message itself.
 718  */
 719 static void
 720 handle_local_ack (void *cls,
 721                   const struct GNUNET_CADET_LocalAck *message)
 722 {
 723   struct GNUNET_CADET_Handle *h = cls;
 724   struct GNUNET_CADET_Channel *ch;
 725
 726   ch = find_channel (h,
 727                      message->ccn);
 728   if (NULL == ch)
 729   {
 730     LOG (GNUNET_ERROR_TYPE_DEBUG,
 731          "ACK on unknown channel %X\n",
 732          ntohl (message->ccn.channel_of_client));
 733     return;
 734   }
 735   ch->allow_send++;
 736   LOG (GNUNET_ERROR_TYPE_DEBUG,
 737        "Got an ACK on mq channel %X (peer %s); new window size is %u!\n",
 738        ntohl (ch->ccn.channel_of_client),
 739        GNUNET_i2s (&ch->peer),
 740        ch->allow_send);
 741   if (NULL == ch->pending_env)
 742   {
 743     LOG (GNUNET_ERROR_TYPE_DEBUG,
 744          "Got an ACK on mq channel %X, allow send now %u!\n",
 745          ntohl (ch->ccn.channel_of_client),
 746          ch->allow_send);
 747     notify_window_size (ch);
 748     return;
 749   }
 750   if (NULL != ch->mq_cont)
 751     return; /* already working on it! */
 752   ch->mq_cont
 753     = GNUNET_SCHEDULER_add_now (&cadet_mq_send_now,
 754                                 ch);
 755 }
 756
 757
 758 /**
 759  * Function called during #GNUNET_CADET_disconnect() to destroy
 760  * all channels that are still open.
 761  *
 762  * @param cls the `struct GNUNET_CADET_Handle`
 763  * @param cid chanenl ID
 764  * @param value a `struct GNUNET_CADET_Channel` to destroy
 765  * @return #GNUNET_OK (continue to iterate)
 766  */
 767 static int
 768 destroy_channel_cb (void *cls,
 769                     uint32_t cid,
 770                     void *value)
 771 {
 772   /* struct GNUNET_CADET_Handle *handle = cls; */
 773   struct GNUNET_CADET_Channel *ch = value;
 774
 775   (void) cls;
 776   (void) cid;
 777   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 778               "Destroying channel due to GNUNET_CADET_disconnect()\n");
 779   destroy_channel (ch);
 780   return GNUNET_OK;
 781 }
 782
 783
 784 /**
 785  * Generic error handler, called with the appropriate error code and
 786  * the same closure specified at the creation of the message queue.
 787  * Not every message queue implementation supports an error handler.
 788  *
 789  * @param cls closure, a `struct GNUNET_CORE_Handle *`
 790  * @param error error code
 791  */
 792 static void
 793 handle_mq_error (void *cls,
 794                  enum GNUNET_MQ_Error error)
 795 {
 796   struct GNUNET_CADET_Handle *h = cls;
 797
 798   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 799               "MQ ERROR: %u\n",
 800               error);
 801   GNUNET_CONTAINER_multihashmap32_iterate (h->channels,
 802                                            &destroy_channel_cb,
 803                                            h);
 804   GNUNET_MQ_destroy (h->mq);
 805   h->mq = NULL;
 806   GNUNET_assert (NULL == h->reconnect_task);
 807   h->reconnect_task = GNUNET_SCHEDULER_add_delayed (h->reconnect_time,
 808                                                     &reconnect_cbk,
 809                                                     h);
 810 }
 811
 812
 813 /**
 814  * Reconnect to the service, retransmit all infomation to try to restore the
 815  * original state.
 816  *
 817  * @param h handle to the cadet
 818  */
 819 static void
 820 reconnect (struct GNUNET_CADET_Handle *h)
 821 {
 822   struct GNUNET_MQ_MessageHandler handlers[] = {
 823     GNUNET_MQ_hd_fixed_size (channel_created,
 824                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE,
 825                              struct GNUNET_CADET_LocalChannelCreateMessage,
 826                              h),
 827     GNUNET_MQ_hd_fixed_size (channel_destroy,
 828                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY,
 829                              struct GNUNET_CADET_LocalChannelDestroyMessage,
 830                              h),
 831     GNUNET_MQ_hd_var_size (local_data,
 832                            GNUNET_MESSAGE_TYPE_CADET_LOCAL_DATA,
 833                            struct GNUNET_CADET_LocalData,
 834                            h),
 835     GNUNET_MQ_hd_fixed_size (local_ack,
 836                              GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK,
 837                              struct GNUNET_CADET_LocalAck,
 838                              h),
 839     GNUNET_MQ_handler_end ()
 840   };
 841
 842   GNUNET_assert (NULL == h->mq);
 843   h->mq = GNUNET_CLIENT_connect (h->cfg,
 844                                  "cadet",
 845                                  handlers,
 846                                  &handle_mq_error,
 847                                  h);
 848 }
 849
 850
 851 /**
 852  * Function called during #GNUNET_CADET_disconnect() to destroy
 853  * all ports that are still open.
 854  *
 855  * @param cls the `struct GNUNET_CADET_Handle`
 856  * @param id port ID
 857  * @param value a `struct GNUNET_CADET_Channel` to destroy
 858  * @return #GNUNET_OK (continue to iterate)
 859  */
 860 static int
 861 destroy_port_cb (void *cls,
 862                  const struct GNUNET_HashCode *id,
 863                  void *value)
 864 {
 865   /* struct GNUNET_CADET_Handle *handle = cls; */
 866   struct GNUNET_CADET_Port *port = value;
 867
 868   (void) cls;
 869   (void) id;
 870   /* This is a warning, the app should have cleanly closed all open ports */
 871   GNUNET_break (0);
 872   GNUNET_CADET_close_port (port);
 873   return GNUNET_OK;
 874 }
 875
 876
 877 /**
 878  * Disconnect from the cadet service. All channels will be destroyed. All channel
 879  * disconnect callbacks will be called on any still connected peers, notifying
 880  * about their disconnection. The registered inbound channel cleaner will be
 881  * called should any inbound channels still exist.
 882  *
 883  * @param handle connection to cadet to disconnect
 884  */
 885 void
 886 GNUNET_CADET_disconnect (struct GNUNET_CADET_Handle *handle)
 887 {
 888   GNUNET_CONTAINER_multihashmap_iterate (handle->ports,
 889                                          &destroy_port_cb,
 890                                          handle);
 891   GNUNET_CONTAINER_multihashmap_destroy (handle->ports);
 892   handle->ports = NULL;
 893   GNUNET_CONTAINER_multihashmap32_iterate (handle->channels,
 894                                            &destroy_channel_cb,
 895                                            handle);
 896   GNUNET_CONTAINER_multihashmap32_destroy (handle->channels);
 897   handle->channels = NULL;
 898   if (NULL != handle->mq)
 899   {
 900     GNUNET_MQ_destroy (handle->mq);
 901     handle->mq = NULL;
 902   }
 903   if (NULL != handle->reconnect_task)
 904   {
 905     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 906     handle->reconnect_task = NULL;
 907   }
 908   GNUNET_free (handle);
 909 }
 910
 911
 912 /**
 913  * Close a port opened with @a GNUNET_CADET_open_port().
 914  * The @a new_channel callback will no longer be called.
 915  *
 916  * @param p Port handle.
 917  */
 918 void
 919 GNUNET_CADET_close_port (struct GNUNET_CADET_Port *p)
 920 {
 921   GNUNET_assert (GNUNET_YES ==
 922                  GNUNET_CONTAINER_multihashmap_remove (p->cadet->ports,
 923                                                        &p->id,
 924                                                        p));
 925   if (NULL != p->cadet->mq)
 926   {
 927     struct GNUNET_CADET_PortMessage *msg;
 928     struct GNUNET_MQ_Envelope *env;
 929
 930     env = GNUNET_MQ_msg (msg,
 931                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_PORT_CLOSE);
 932     msg->port = p->id;
 933     GNUNET_MQ_send (p->cadet->mq,
 934                     env);
 935   }
 936   GNUNET_free_non_null (p->handlers);
 937   GNUNET_free (p);
 938 }
 939
 940
 941 /**
 942  * Destroy an existing channel.
 943  *
 944  * The existing end callback for the channel will NOT be called.
 945  * Any pending outgoing messages will be sent but no incoming messages will be
 946  * accepted and no data callbacks will be called.
 947  *
 948  * @param channel Channel handle, becomes invalid after this call.
 949  */
 950 void
 951 GNUNET_CADET_channel_destroy (struct GNUNET_CADET_Channel *channel)
 952 {
 953   struct GNUNET_CADET_Handle *h = channel->cadet;
 954   struct GNUNET_CADET_LocalChannelDestroyMessage *msg;
 955   struct GNUNET_MQ_Envelope *env;
 956
 957   if (NULL != h->mq)
 958   {
 959     env = GNUNET_MQ_msg (msg,
 960                          GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_DESTROY);
 961     msg->ccn = channel->ccn;
 962     GNUNET_MQ_send (h->mq,
 963                     env);
 964   }
 965   GNUNET_log (GNUNET_ERROR_TYPE_INFO,
 966               "Destroying channel due to GNUNET_CADET_channel_destroy()\n");
 967   channel->disconnects = NULL;
 968   destroy_channel (channel);
 969 }
 970
 971
 972 /**
 973  * Get information about a channel.
 974  *
 975  * @param channel Channel handle.
 976  * @param option Query (GNUNET_CADET_OPTION_*).
 977  * @param ... dependant on option, currently not used
 978  *
 979  * @return Union with an answer to the query.
 980  */
 981 const union GNUNET_CADET_ChannelInfo *
 982 GNUNET_CADET_channel_get_info (struct GNUNET_CADET_Channel *channel,
 983                                enum GNUNET_CADET_ChannelOption option,
 984                                ...)
 985 {
 986   static int bool_flag;
 987
 988   switch (option)
 989   {
 990     case GNUNET_CADET_OPTION_NOBUFFER:
 991     case GNUNET_CADET_OPTION_RELIABLE:
 992     case GNUNET_CADET_OPTION_OUT_OF_ORDER:
 993       if (0 != (option & channel->options))
 994         bool_flag = GNUNET_YES;
 995       else
 996         bool_flag = GNUNET_NO;
 997       return (const union GNUNET_CADET_ChannelInfo *) &bool_flag;
 998       break;
 999     case GNUNET_CADET_OPTION_PEER:
1000       return (const union GNUNET_CADET_ChannelInfo *) &channel->peer;
1001       break;
1002     default:
1003       GNUNET_break (0);
1004       return NULL;
1005   }
1006 }
1007
1008
1009 /**
1010  * Send an ack on the channel to confirm the processing of a message.
1011  *
1012  * @param ch Channel on which to send the ACK.
1013  */
1014 void
1015 GNUNET_CADET_receive_done (struct GNUNET_CADET_Channel *channel)
1016 {
1017   struct GNUNET_CADET_LocalAck *msg;
1018   struct GNUNET_MQ_Envelope *env;
1019
1020   env = GNUNET_MQ_msg (msg,
1021                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_ACK);
1022   LOG (GNUNET_ERROR_TYPE_DEBUG,
1023        "Sending ACK on channel %X\n",
1024        ntohl (channel->ccn.channel_of_client));
1025   msg->ccn = channel->ccn;
1026   GNUNET_MQ_send (channel->cadet->mq,
1027                   env);
1028 }
1029
1030
1031 /**
1032  * Connect to the MQ-based cadet service.
1033  *
1034  * @param cfg Configuration to use.
1035  *
1036  * @return Handle to the cadet service NULL on error.
1037  */
1038 struct GNUNET_CADET_Handle *
1039 GNUNET_CADET_connect (const struct GNUNET_CONFIGURATION_Handle *cfg)
1040 {
1041   struct GNUNET_CADET_Handle *h;
1042
1043   LOG (GNUNET_ERROR_TYPE_DEBUG,
1044        "GNUNET_CADET_connect()\n");
1045   h = GNUNET_new (struct GNUNET_CADET_Handle);
1046   h->cfg = cfg;
1047   h->ports = GNUNET_CONTAINER_multihashmap_create (4,
1048                                                    GNUNET_YES);
1049   h->channels = GNUNET_CONTAINER_multihashmap32_create (4);
1050   reconnect (h);
1051   if (NULL == h->mq)
1052   {
1053     GNUNET_break (0);
1054     GNUNET_CADET_disconnect (h);
1055     return NULL;
1056   }
1057   h->next_ccn.channel_of_client = htonl (GNUNET_CADET_LOCAL_CHANNEL_ID_CLI);
1058   return h;
1059 }
1060
1061
1062 /**
1063  * Open a port to receive incomming MQ-based channels.
1064  *
1065  * @param h CADET handle.
1066  * @param port Hash identifying the port.
1067  * @param connects Function called when an incoming channel is connected.
1068  * @param connects_cls Closure for the @a connects handler.
1069  * @param window_changes Function called when the transmit window size changes.
1070  * @param disconnects Function called when a channel is disconnected.
1071  * @param handlers Callbacks for messages we care about, NULL-terminated.
1072  * @return Port handle, NULL if port is in use
1073  */
1074 struct GNUNET_CADET_Port *
1075 GNUNET_CADET_open_port (struct GNUNET_CADET_Handle *h,
1076                         const struct GNUNET_HashCode *port,
1077                         GNUNET_CADET_ConnectEventHandler connects,
1078                         void * connects_cls,
1079                         GNUNET_CADET_WindowSizeEventHandler window_changes,
1080                         GNUNET_CADET_DisconnectEventHandler disconnects,
1081                         const struct GNUNET_MQ_MessageHandler *handlers)
1082 {
1083   struct GNUNET_CADET_Port *p;
1084
1085   GNUNET_assert (NULL != connects);
1086   GNUNET_assert (NULL != disconnects);
1087   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1088               "Listening to CADET port %s\n",
1089               GNUNET_h2s (port));
1090
1091   p = GNUNET_new (struct GNUNET_CADET_Port);
1092   p->cadet = h;
1093   p->id = *port;
1094   if (GNUNET_OK !=
1095       GNUNET_CONTAINER_multihashmap_put (h->ports,
1096                                          &p->id,
1097                                          p,
1098                                          GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY))
1099   {
1100     GNUNET_free (p);
1101     return NULL;
1102   }
1103   p->connects = connects;
1104   p->cls = connects_cls;
1105   p->window_changes = window_changes;
1106   p->disconnects = disconnects;
1107   p->handlers = GNUNET_MQ_copy_handlers (handlers);
1108
1109   GNUNET_assert (GNUNET_OK ==
1110                  open_port_cb (h,
1111                                &p->id,
1112                                p));
1113   return p;
1114 }
1115
1116
1117 /**
1118  * Create a new channel towards a remote peer.
1119  *
1120  * If the destination port is not open by any peer or the destination peer
1121  * does not accept the channel, #GNUNET_CADET_ChannelEndHandler will be called
1122  * for this channel.
1123  *
1124  * @param h CADET handle.
1125  * @param channel_cls Closure for the channel. It's given to:
1126  *                    - The disconnect handler @a disconnects
1127  *                    - Each message type callback in @a handlers
1128  * @param destination Peer identity the channel should go to.
1129  * @param port Identification of the destination port.
1130  * @param options CadetOption flag field, with all desired option bits set to 1.
1131  * @param window_changes Function called when the transmit window size changes.
1132  * @param disconnects Function called when the channel is disconnected.
1133  * @param handlers Callbacks for messages we care about, NULL-terminated.
1134  * @return Handle to the channel.
1135  */
1136 struct GNUNET_CADET_Channel *
1137 GNUNET_CADET_channel_create (struct GNUNET_CADET_Handle *h,
1138                              void *channel_cls,
1139                              const struct GNUNET_PeerIdentity *destination,
1140                              const struct GNUNET_HashCode *port,
1141                              enum GNUNET_CADET_ChannelOption options,
1142                              GNUNET_CADET_WindowSizeEventHandler window_changes,
1143                              GNUNET_CADET_DisconnectEventHandler disconnects,
1144                              const struct GNUNET_MQ_MessageHandler *handlers)
1145 {
1146   struct GNUNET_CADET_Channel *ch;
1147   struct GNUNET_CADET_LocalChannelCreateMessage *msg;
1148   struct GNUNET_MQ_Envelope *env;
1149
1150   GNUNET_assert (NULL != disconnects);
1151   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
1152               "Creating channel to peer %s at port %s\n",
1153               GNUNET_i2s (destination),
1154               GNUNET_h2s (port));
1155   ch = create_channel (h,
1156                        NULL);
1157   ch->ctx = channel_cls;
1158   ch->peer = *destination;
1159   ch->options = options;
1160   ch->window_changes = window_changes;
1161   ch->disconnects = disconnects;
1162
1163   /* Create MQ for channel */
1164   ch->mq = GNUNET_MQ_queue_for_callbacks (&cadet_mq_send_impl,
1165                                           &cadet_mq_destroy_impl,
1166                                           &cadet_mq_cancel_impl,
1167                                           ch,
1168                                           handlers,
1169                                           &cadet_mq_error_handler,
1170                                           ch);
1171   GNUNET_MQ_set_handlers_closure (ch->mq,
1172                                   channel_cls);
1173
1174   /* Request channel creation to service */
1175   env = GNUNET_MQ_msg (msg,
1176                        GNUNET_MESSAGE_TYPE_CADET_LOCAL_CHANNEL_CREATE);
1177   msg->ccn = ch->ccn;
1178   msg->port = *port;
1179   msg->peer = *destination;
1180   msg->opt = htonl (options);
1181   GNUNET_MQ_send (h->mq,
1182                   env);
1183   return ch;
1184 }
1185
1186
1187 /**
1188  * Obtain the message queue for a connected peer.
1189  *
1190  * @param channel The channel handle from which to get the MQ.
1191  *
1192  * @return NULL if @a channel is not yet connected.
1193  */
1194 struct GNUNET_MQ_Handle *
1195 GNUNET_CADET_get_mq (const struct GNUNET_CADET_Channel *channel)
1196 {
1197   return channel->mq;
1198 }
1199
1200 /* end of cadet_api.c */