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