src/core/core_api.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2009-2016 GNUnet e.V.
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 51 Franklin Street, Fifth Floor,
  18      Boston, MA 02110-1301, USA.
  19 */
  20 /**
  21  * @file core/core_api.c
  22  * @brief core service; this is the main API for encrypted P2P
  23  *        communications
  24  * @author Christian Grothoff
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_constants.h"
  29 #include "gnunet_core_service.h"
  30 #include "core.h"
  31
  32 #define LOG(kind,...) GNUNET_log_from (kind, "core-api",__VA_ARGS__)
  33
  34
  35 /**
  36  * Handle for a transmission request.
  37  */
  38 struct GNUNET_CORE_TransmitHandle
  39 {
  40
  41   /**
  42    * Corresponding peer record.
  43    */
  44   struct PeerRecord *peer;
  45
  46   /**
  47    * Function that will be called to get the actual request
  48    * (once we are ready to transmit this request to the core).
  49    * The function will be called with a NULL buffer to signal
  50    * timeout.
  51    */
  52   GNUNET_CONNECTION_TransmitReadyNotify get_message;
  53
  54   /**
  55    * Closure for @e get_message.
  56    */
  57   void *get_message_cls;
  58
  59   /**
  60    * Deadline for the transmission (the request does not get cancelled
  61    * at this time, this is merely how soon the application wants this out).
  62    */
  63   struct GNUNET_TIME_Absolute deadline;
  64
  65   /**
  66    * When did this request get queued?
  67    */
  68   struct GNUNET_TIME_Absolute request_time;
  69
  70   /**
  71    * How important is this message?
  72    */
  73   enum GNUNET_CORE_Priority priority;
  74
  75   /**
  76    * Is corking allowed?
  77    */
  78   int cork;
  79
  80   /**
  81    * Size of this request.
  82    */
  83   uint16_t msize;
  84
  85   /**
  86    * Send message request ID for this request.
  87    */
  88   uint16_t smr_id;
  89
  90 };
  91
  92
  93 /**
  94  * Information we track for each peer.
  95  */
  96 struct PeerRecord
  97 {
  98
  99   /**
 100    * Corresponding CORE handle.
 101    */
 102   struct GNUNET_CORE_Handle *ch;
 103
 104   /**
 105    * Pending request, if any. 'th->peer' is set to NULL if the
 106    * request is not active.
 107    */
 108   struct GNUNET_CORE_TransmitHandle th;
 109
 110   /**
 111    * Peer the record is about.
 112    */
 113   struct GNUNET_PeerIdentity peer;
 114
 115   /**
 116    * SendMessageRequest ID generator for this peer.
 117    */
 118   uint16_t smr_id_gen;
 119
 120 };
 121
 122
 123 /**
 124  * Context for the core service connection.
 125  */
 126 struct GNUNET_CORE_Handle
 127 {
 128
 129   /**
 130    * Configuration we're using.
 131    */
 132   const struct GNUNET_CONFIGURATION_Handle *cfg;
 133
 134   /**
 135    * Closure for the various callbacks.
 136    */
 137   void *cls;
 138
 139   /**
 140    * Function to call once we've handshaked with the core service.
 141    */
 142   GNUNET_CORE_StartupCallback init;
 143
 144   /**
 145    * Function to call whenever we're notified about a peer connecting.
 146    */
 147   GNUNET_CORE_ConnectEventHandler connects;
 148
 149   /**
 150    * Function to call whenever we're notified about a peer disconnecting.
 151    */
 152   GNUNET_CORE_DisconnectEventHandler disconnects;
 153
 154   /**
 155    * Function to call whenever we receive an inbound message.
 156    */
 157   GNUNET_CORE_MessageCallback inbound_notify;
 158
 159   /**
 160    * Function to call whenever we receive an outbound message.
 161    */
 162   GNUNET_CORE_MessageCallback outbound_notify;
 163
 164   /**
 165    * Function handlers for messages of particular type.
 166    */
 167   struct GNUNET_CORE_MessageHandler *handlers;
 168
 169   /**
 170    * Our message queue for transmissions to the service.
 171    */
 172   struct GNUNET_MQ_Handle *mq;
 173
 174   /**
 175    * Hash map listing all of the peers that we are currently
 176    * connected to.
 177    */
 178   struct GNUNET_CONTAINER_MultiPeerMap *peers;
 179
 180   /**
 181    * Identity of this peer.
 182    */
 183   struct GNUNET_PeerIdentity me;
 184
 185   /**
 186    * ID of reconnect task (if any).
 187    */
 188   struct GNUNET_SCHEDULER_Task *reconnect_task;
 189
 190   /**
 191    * Current delay we use for re-trying to connect to core.
 192    */
 193   struct GNUNET_TIME_Relative retry_backoff;
 194
 195   /**
 196    * Number of entries in the handlers array.
 197    */
 198   unsigned int hcnt;
 199
 200   /**
 201    * For inbound notifications without a specific handler, do
 202    * we expect to only receive headers?
 203    */
 204   int inbound_hdr_only;
 205
 206   /**
 207    * For outbound notifications without a specific handler, do
 208    * we expect to only receive headers?
 209    */
 210   int outbound_hdr_only;
 211
 212   /**
 213    * Are we currently disconnected and hence unable to forward
 214    * requests?
 215    */
 216   int currently_down;
 217
 218 };
 219
 220
 221 /**
 222  * Our current client connection went down.  Clean it up
 223  * and try to reconnect!
 224  *
 225  * @param h our handle to the core service
 226  */
 227 static void
 228 reconnect (struct GNUNET_CORE_Handle *h);
 229
 230
 231 /**
 232  * Task schedule to try to re-connect to core.
 233  *
 234  * @param cls the `struct GNUNET_CORE_Handle`
 235  * @param tc task context
 236  */
 237 static void
 238 reconnect_task (void *cls)
 239 {
 240   struct GNUNET_CORE_Handle *h = cls;
 241
 242   h->reconnect_task = NULL;
 243   LOG (GNUNET_ERROR_TYPE_DEBUG,
 244        "Connecting to CORE service after delay\n");
 245   reconnect (h);
 246 }
 247
 248
 249 /**
 250  * Notify clients about disconnect and free the entry for connected
 251  * peer.
 252  *
 253  * @param cls the `struct GNUNET_CORE_Handle *`
 254  * @param key the peer identity (not used)
 255  * @param value the `struct PeerRecord` to free.
 256  * @return #GNUNET_YES (continue)
 257  */
 258 static int
 259 disconnect_and_free_peer_entry (void *cls,
 260                                 const struct GNUNET_PeerIdentity *key,
 261                                 void *value)
 262 {
 263   struct GNUNET_CORE_Handle *h = cls;
 264   struct GNUNET_CORE_TransmitHandle *th;
 265   struct PeerRecord *pr = value;
 266
 267   if (NULL != h->disconnects)
 268     h->disconnects (h->cls,
 269                     &pr->peer);
 270   /* all requests should have been cancelled, clean up anyway, just in case */
 271   th = &pr->th;
 272   if (NULL != th->peer)
 273   {
 274     GNUNET_break (0);
 275     th->peer = NULL;
 276   }
 277   /* done with 'voluntary' cleanups, now on to normal freeing */
 278   GNUNET_assert (GNUNET_YES ==
 279                  GNUNET_CONTAINER_multipeermap_remove (h->peers,
 280                                                        key,
 281                                                        pr));
 282   GNUNET_assert (pr->ch == h);
 283   GNUNET_free (pr);
 284   return GNUNET_YES;
 285 }
 286
 287
 288 /**
 289  * Close down any existing connection to the CORE service and
 290  * try re-establishing it later.
 291  *
 292  * @param h our handle
 293  */
 294 static void
 295 reconnect_later (struct GNUNET_CORE_Handle *h)
 296 {
 297   GNUNET_assert (NULL == h->reconnect_task);
 298   if (NULL != h->mq)
 299   {
 300     GNUNET_MQ_destroy (h->mq);
 301     h->mq = NULL;
 302   }
 303   h->currently_down = GNUNET_YES;
 304   GNUNET_assert (h->reconnect_task == NULL);
 305   h->reconnect_task =
 306       GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
 307                                     &reconnect_task,
 308                                     h);
 309   GNUNET_CONTAINER_multipeermap_iterate (h->peers,
 310                                          &disconnect_and_free_peer_entry,
 311                                          h);
 312   h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
 313 }
 314
 315
 316 /**
 317  * Generic error handler, called with the appropriate error code and
 318  * the same closure specified at the creation of the message queue.
 319  * Not every message queue implementation supports an error handler.
 320  *
 321  * @param cls closure, a `struct GNUNET_CORE_Handle *`
 322  * @param error error code
 323  */
 324 static void
 325 handle_mq_error (void *cls,
 326                  enum GNUNET_MQ_Error error)
 327 {
 328   struct GNUNET_CORE_Handle *h = cls;
 329
 330   reconnect_later (h);
 331 }
 332
 333
 334 /**
 335  * Handle  init  reply message  received  from  CORE service.   Notify
 336  * application  that we  are now  connected  to the  CORE.  Also  fake
 337  * loopback connection.
 338  *
 339  * @param cls the `struct GNUNET_CORE_Handle`
 340  * @param m the init reply
 341  */
 342 static void
 343 handle_init_reply (void *cls,
 344                    const struct InitReplyMessage *m)
 345 {
 346   struct GNUNET_CORE_Handle *h = cls;
 347   GNUNET_CORE_StartupCallback init;
 348   struct PeerRecord *pr;
 349
 350   GNUNET_break (0 == ntohl (m->reserved));
 351   GNUNET_break (GNUNET_YES == h->currently_down);
 352   h->currently_down = GNUNET_NO;
 353   h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 354   if (NULL != (init = h->init))
 355   {
 356     /* mark so we don't call init on reconnect */
 357     h->init = NULL;
 358     h->me = m->my_identity;
 359     LOG (GNUNET_ERROR_TYPE_DEBUG,
 360          "Connected to core service of peer `%s'.\n",
 361          GNUNET_i2s (&h->me));
 362     init (h->cls,
 363           &h->me);
 364   }
 365   else
 366   {
 367     LOG (GNUNET_ERROR_TYPE_DEBUG,
 368          "Successfully reconnected to core service.\n");
 369     GNUNET_break (0 == memcmp (&h->me,
 370                                &m->my_identity,
 371                                sizeof (struct GNUNET_PeerIdentity)));
 372   }
 373   /* fake 'connect to self' */
 374   pr = GNUNET_new (struct PeerRecord);
 375   pr->peer = h->me;
 376   pr->ch = h;
 377   GNUNET_assert (GNUNET_YES ==
 378                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 379                                                     &h->me,
 380                                                     pr,
 381                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 382   if (NULL != h->connects)
 383     h->connects (h->cls,
 384                  &pr->peer);
 385 }
 386
 387
 388 /**
 389  * Handle connect message received from CORE service.
 390  * Notify the application about the new connection.
 391  *
 392  * @param cls the `struct GNUNET_CORE_Handle`
 393  * @param cnm the connect message
 394  */
 395 static void
 396 handle_connect_notify (void *cls,
 397                        const struct ConnectNotifyMessage * cnm)
 398 {
 399   struct GNUNET_CORE_Handle *h = cls;
 400   struct PeerRecord *pr;
 401
 402   GNUNET_break (GNUNET_NO == h->currently_down);
 403   LOG (GNUNET_ERROR_TYPE_DEBUG,
 404        "Received notification about connection from `%s'.\n",
 405        GNUNET_i2s (&cnm->peer));
 406   if (0 == memcmp (&h->me,
 407                    &cnm->peer,
 408                    sizeof (struct GNUNET_PeerIdentity)))
 409   {
 410     /* connect to self!? */
 411     GNUNET_break (0);
 412     return;
 413   }
 414   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 415                                           &cnm->peer);
 416   if (NULL != pr)
 417   {
 418     GNUNET_break (0);
 419     reconnect_later (h);
 420     return;
 421   }
 422   pr = GNUNET_new (struct PeerRecord);
 423   pr->peer = cnm->peer;
 424   pr->ch = h;
 425   GNUNET_assert (GNUNET_YES ==
 426                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 427                                                     &cnm->peer,
 428                                                     pr,
 429                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 430   if (NULL != h->connects)
 431     h->connects (h->cls,
 432                  &pr->peer);
 433 }
 434
 435
 436 /**
 437  * Handle disconnect message received from CORE service.
 438  * Notify the application about the lost connection.
 439  *
 440  * @param cls the `struct GNUNET_CORE_Handle`
 441  * @param dnm message about the disconnect event
 442  */
 443 static void
 444 handle_disconnect_notify (void *cls,
 445                           const struct DisconnectNotifyMessage * dnm)
 446 {
 447   struct GNUNET_CORE_Handle *h = cls;
 448   struct PeerRecord *pr;
 449
 450   GNUNET_break (GNUNET_NO == h->currently_down);
 451   if (0 == memcmp (&h->me,
 452                    &dnm->peer,
 453                    sizeof (struct GNUNET_PeerIdentity)))
 454   {
 455     /* connection to self!? */
 456     GNUNET_break (0);
 457     return;
 458   }
 459   GNUNET_break (0 == ntohl (dnm->reserved));
 460   LOG (GNUNET_ERROR_TYPE_DEBUG,
 461        "Received notification about disconnect from `%s'.\n",
 462        GNUNET_i2s (&dnm->peer));
 463   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 464                                           &dnm->peer);
 465   if (NULL == pr)
 466   {
 467     GNUNET_break (0);
 468     reconnect_later (h);
 469     return;
 470   }
 471   disconnect_and_free_peer_entry (h,
 472                                   &dnm->peer,
 473                                   pr);
 474 }
 475
 476
 477 /**
 478  * Check that message received from CORE service is well-formed.
 479  *
 480  * @param cls the `struct GNUNET_CORE_Handle`
 481  * @param ntm the message we got
 482  * @return #GNUNET_OK if the message is well-formed
 483  */
 484 static int
 485 check_notify_inbound (void *cls,
 486                       const struct NotifyTrafficMessage *ntm)
 487 {
 488   struct GNUNET_CORE_Handle *h = cls;
 489   uint16_t msize;
 490   const struct GNUNET_MessageHeader *em;
 491
 492   GNUNET_break (GNUNET_NO == h->currently_down);
 493   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 494   if (msize < sizeof (struct GNUNET_MessageHeader))
 495   {
 496     GNUNET_break (0);
 497     return GNUNET_SYSERR;
 498   }
 499   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 500   if ( (GNUNET_NO == h->inbound_hdr_only) &&
 501        (msize != ntohs (em->size)) )
 502   {
 503     GNUNET_break (0);
 504     return GNUNET_SYSERR;
 505   }
 506   return GNUNET_OK;
 507 }
 508
 509
 510 /**
 511  * Handle inbound message received from CORE service.  If applicable,
 512  * notify the application.
 513  *
 514  * @param cls the `struct GNUNET_CORE_Handle`
 515  * @param ntm the message we got from CORE.
 516  */
 517 static void
 518 handle_notify_inbound (void *cls,
 519                        const struct NotifyTrafficMessage *ntm)
 520 {
 521   struct GNUNET_CORE_Handle *h = cls;
 522   const struct GNUNET_MessageHeader *em;
 523   struct PeerRecord *pr;
 524   uint16_t et;
 525
 526   GNUNET_break (GNUNET_NO == h->currently_down);
 527   LOG (GNUNET_ERROR_TYPE_DEBUG,
 528        "Received inbound message from `%s'.\n",
 529        GNUNET_i2s (&ntm->peer));
 530   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 531   et = ntohs (em->type);
 532   for (unsigned int hpos = 0; NULL != h->handlers[hpos].callback; hpos++)
 533   {
 534     const struct GNUNET_CORE_MessageHandler *mh;
 535
 536     mh = &h->handlers[hpos];
 537     if (mh->type != et)
 538       continue;
 539     if ( (mh->expected_size != ntohs (em->size)) &&
 540          (0 != mh->expected_size) )
 541     {
 542       LOG (GNUNET_ERROR_TYPE_ERROR,
 543            "Unexpected message size %u for message of type %u from peer `%s'\n",
 544            htons (em->size),
 545            mh->type,
 546            GNUNET_i2s (&ntm->peer));
 547       GNUNET_break_op (0);
 548       continue;
 549     }
 550     pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 551                                             &ntm->peer);
 552     if (NULL == pr)
 553     {
 554       GNUNET_break (0);
 555       reconnect_later (h);
 556       return;
 557     }
 558     if (GNUNET_OK !=
 559         h->handlers[hpos].callback (h->cls,
 560                                     &ntm->peer,
 561                                     em))
 562     {
 563       /* error in processing, do not process other messages! */
 564       break;
 565     }
 566   }
 567   if (NULL != h->inbound_notify)
 568     h->inbound_notify (h->cls,
 569                        &ntm->peer,
 570                        em);
 571 }
 572
 573
 574 /**
 575  * Check that message received from CORE service is well-formed.
 576  *
 577  * @param cls the `struct GNUNET_CORE_Handle`
 578  * @param ntm the message we got
 579  * @return #GNUNET_OK if the message is well-formed
 580  */
 581 static int
 582 check_notify_outbound (void *cls,
 583                        const struct NotifyTrafficMessage *ntm)
 584 {
 585   struct GNUNET_CORE_Handle *h = cls;
 586   uint16_t msize;
 587   const struct GNUNET_MessageHeader *em;
 588
 589   GNUNET_break (GNUNET_NO == h->currently_down);
 590   LOG (GNUNET_ERROR_TYPE_DEBUG,
 591        "Received outbound message from `%s'.\n",
 592        GNUNET_i2s (&ntm->peer));
 593   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 594   if (msize < sizeof (struct GNUNET_MessageHeader))
 595   {
 596     GNUNET_break (0);
 597     return GNUNET_SYSERR;
 598   }
 599   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 600   if ( (GNUNET_NO == h->outbound_hdr_only) &&
 601        (msize != ntohs (em->size)) )
 602   {
 603     GNUNET_break (0);
 604     return GNUNET_SYSERR;
 605   }
 606   return GNUNET_OK;
 607 }
 608
 609
 610 /**
 611  * Handle outbound message received from CORE service.  If applicable,
 612  * notify the application.
 613  *
 614  * @param cls the `struct GNUNET_CORE_Handle`
 615  * @param ntm the message we got
 616  */
 617 static void
 618 handle_notify_outbound (void *cls,
 619                         const struct NotifyTrafficMessage *ntm)
 620 {
 621   struct GNUNET_CORE_Handle *h = cls;
 622   const struct GNUNET_MessageHeader *em;
 623
 624   GNUNET_break (GNUNET_NO == h->currently_down);
 625   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 626   LOG (GNUNET_ERROR_TYPE_DEBUG,
 627        "Received notification about transmission to `%s'.\n",
 628        GNUNET_i2s (&ntm->peer));
 629   if (NULL == h->outbound_notify)
 630   {
 631     GNUNET_break (0);
 632     return;
 633   }
 634   h->outbound_notify (h->cls,
 635                       &ntm->peer,
 636                       em);
 637 }
 638
 639
 640 /**
 641  * Handle message received from CORE service notifying us that we are
 642  * now allowed to send a message to a peer.  If that message is still
 643  * pending, put it into the queue to be transmitted.
 644  *
 645  * @param cls the `struct GNUNET_CORE_Handle`
 646  * @param ntm the message we got
 647  */
 648 static void
 649 handle_send_ready (void *cls,
 650                    const struct SendMessageReady *smr)
 651 {
 652   struct GNUNET_CORE_Handle *h = cls;
 653   struct PeerRecord *pr;
 654   struct GNUNET_CORE_TransmitHandle *th;
 655   struct SendMessage *sm;
 656   struct GNUNET_MQ_Envelope *env;
 657   struct GNUNET_TIME_Relative delay;
 658   struct GNUNET_TIME_Relative overdue;
 659   unsigned int ret;
 660
 661   GNUNET_break (GNUNET_NO == h->currently_down);
 662   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 663                                           &smr->peer);
 664   if (NULL == pr)
 665   {
 666     GNUNET_break (0);
 667     reconnect_later (h);
 668     return;
 669   }
 670   LOG (GNUNET_ERROR_TYPE_DEBUG,
 671        "Received notification about transmission readiness to `%s'.\n",
 672        GNUNET_i2s (&smr->peer));
 673   if (NULL == pr->th.peer)
 674   {
 675     /* request must have been cancelled between the original request
 676      * and the response from CORE, ignore CORE's readiness */
 677     return;
 678   }
 679   th = &pr->th;
 680   if (ntohs (smr->smr_id) != th->smr_id)
 681   {
 682     /* READY message is for expired or cancelled message,
 683      * ignore! (we should have already sent another request) */
 684     return;
 685   }
 686   /* ok, all good, send message out! */
 687   th->peer = NULL;
 688   env = GNUNET_MQ_msg_extra (sm,
 689                              th->msize,
 690                              GNUNET_MESSAGE_TYPE_CORE_SEND);
 691   sm->priority = htonl ((uint32_t) th->priority);
 692   sm->deadline = GNUNET_TIME_absolute_hton (th->deadline);
 693   sm->peer = pr->peer;
 694   sm->cork = htonl ((uint32_t) th->cork);
 695   sm->reserved = htonl (0);
 696   ret = th->get_message (th->get_message_cls,
 697                          th->msize,
 698                          &sm[1]);
 699   GNUNET_assert (ret == th->msize); /* NOTE: API change! */
 700   delay = GNUNET_TIME_absolute_get_duration (th->request_time);
 701   overdue = GNUNET_TIME_absolute_get_duration (th->deadline);
 702   if (overdue.rel_value_us > GNUNET_CONSTANTS_LATENCY_WARN.rel_value_us)
 703     LOG (GNUNET_ERROR_TYPE_WARNING,
 704          "Transmitting overdue %u bytes to `%s' at priority %u with %s delay %s\n",
 705          ret,
 706          GNUNET_i2s (&pr->peer),
 707          (unsigned int) th->priority,
 708          GNUNET_STRINGS_relative_time_to_string (delay,
 709                                                  GNUNET_YES),
 710          (th->cork) ? " (corked)" : "");
 711   else
 712     LOG (GNUNET_ERROR_TYPE_DEBUG,
 713          "Transmitting %u bytes to `%s' at priority %u with %s delay %s\n",
 714          ret,
 715          GNUNET_i2s (&pr->peer),
 716          (unsigned int) th->priority,
 717          GNUNET_STRINGS_relative_time_to_string (delay,
 718                                                  GNUNET_YES),
 719          (th->cork) ? " (corked)" : "");
 720   GNUNET_MQ_send (h->mq,
 721                   env);
 722 }
 723
 724
 725 /**
 726  * Our current client connection went down.  Clean it up and try to
 727  * reconnect!
 728  *
 729  * @param h our handle to the core service
 730  */
 731 static void
 732 reconnect (struct GNUNET_CORE_Handle *h)
 733 {
 734   GNUNET_MQ_hd_fixed_size (init_reply,
 735                            GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
 736                            struct InitReplyMessage);
 737   GNUNET_MQ_hd_fixed_size (connect_notify,
 738                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
 739                            struct ConnectNotifyMessage);
 740   GNUNET_MQ_hd_fixed_size (disconnect_notify,
 741                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
 742                            struct DisconnectNotifyMessage);
 743   GNUNET_MQ_hd_var_size (notify_inbound,
 744                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
 745                          struct NotifyTrafficMessage);
 746   GNUNET_MQ_hd_var_size (notify_outbound,
 747                          GNUNET_MESSAGE_TYPE_CORE_NOTIFY_OUTBOUND,
 748                          struct NotifyTrafficMessage);
 749   GNUNET_MQ_hd_fixed_size (send_ready,
 750                            GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
 751                            struct SendMessageReady);
 752  struct GNUNET_MQ_MessageHandler handlers[] = {
 753     make_init_reply_handler (h),
 754     make_connect_notify_handler (h),
 755     make_disconnect_notify_handler (h),
 756     make_notify_inbound_handler (h),
 757     make_notify_outbound_handler (h),
 758     make_send_ready_handler (h),
 759     GNUNET_MQ_handler_end ()
 760   };
 761   struct InitMessage *init;
 762   struct GNUNET_MQ_Envelope *env;
 763   uint32_t opt;
 764   uint16_t *ts;
 765
 766   GNUNET_assert (NULL == h->mq);
 767   GNUNET_assert (GNUNET_YES == h->currently_down);
 768   h->mq = GNUNET_CLIENT_connecT (h->cfg,
 769                                  "core",
 770                                  handlers,
 771                                  &handle_mq_error,
 772                                  h);
 773   if (NULL == h->mq)
 774   {
 775     reconnect_later (h);
 776     return;
 777   }
 778   env = GNUNET_MQ_msg_extra (init,
 779                              sizeof (uint16_t) * h->hcnt,
 780                              GNUNET_MESSAGE_TYPE_CORE_INIT);
 781   opt = 0;
 782   if (NULL != h->inbound_notify)
 783   {
 784     if (h->inbound_hdr_only)
 785       opt |= GNUNET_CORE_OPTION_SEND_HDR_INBOUND;
 786     else
 787       opt |= GNUNET_CORE_OPTION_SEND_FULL_INBOUND;
 788   }
 789   if (NULL != h->outbound_notify)
 790   {
 791     if (h->outbound_hdr_only)
 792       opt |= GNUNET_CORE_OPTION_SEND_HDR_OUTBOUND;
 793     else
 794       opt |= GNUNET_CORE_OPTION_SEND_FULL_OUTBOUND;
 795   }
 796   LOG (GNUNET_ERROR_TYPE_INFO,
 797        "(Re)connecting to CORE service, monitoring messages of type %u\n",
 798        opt);
 799   init->options = htonl (opt);
 800   ts = (uint16_t *) &init[1];
 801   for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
 802     ts[hpos] = htons (h->handlers[hpos].type);
 803   GNUNET_MQ_send (h->mq,
 804                   env);
 805 }
 806
 807
 808 /**
 809  * Connect to the core service.  Note that the connection may complete
 810  * (or fail) asynchronously.
 811  *
 812  * @param cfg configuration to use
 813  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
 814  * @param init callback to call once we have successfully
 815  *        connected to the core service
 816  * @param connects function to call on peer connect, can be NULL
 817  * @param disconnects function to call on peer disconnect / timeout, can be NULL
 818  * @param inbound_notify function to call for all inbound messages, can be NULL
 819  * @param inbound_hdr_only set to #GNUNET_YES if inbound_notify will only read the
 820  *                GNUNET_MessageHeader and hence we do not need to give it the full message;
 821  *                can be used to improve efficiency, ignored if @a inbound_notify is NULLL
 822  * @param outbound_notify function to call for all outbound messages, can be NULL
 823  * @param outbound_hdr_only set to #GNUNET_YES if outbound_notify will only read the
 824  *                GNUNET_MessageHeader and hence we do not need to give it the full message
 825  *                can be used to improve efficiency, ignored if @a outbound_notify is NULLL
 826  * @param handlers callbacks for messages we care about, NULL-terminated
 827  * @return handle to the core service (only useful for disconnect until 'init' is called);
 828  *                NULL on error (in this case, init is never called)
 829  */
 830 struct GNUNET_CORE_Handle *
 831 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 832                      void *cls,
 833                      GNUNET_CORE_StartupCallback init,
 834                      GNUNET_CORE_ConnectEventHandler connects,
 835                      GNUNET_CORE_DisconnectEventHandler disconnects,
 836                      GNUNET_CORE_MessageCallback inbound_notify,
 837                      int inbound_hdr_only,
 838                      GNUNET_CORE_MessageCallback outbound_notify,
 839                      int outbound_hdr_only,
 840                      const struct GNUNET_CORE_MessageHandler *handlers)
 841 {
 842   struct GNUNET_CORE_Handle *h;
 843   unsigned int hcnt;
 844
 845   h = GNUNET_new (struct GNUNET_CORE_Handle);
 846   h->cfg = cfg;
 847   h->cls = cls;
 848   h->init = init;
 849   h->connects = connects;
 850   h->disconnects = disconnects;
 851   h->inbound_notify = inbound_notify;
 852   h->outbound_notify = outbound_notify;
 853   h->inbound_hdr_only = inbound_hdr_only;
 854   h->outbound_hdr_only = outbound_hdr_only;
 855   h->currently_down = GNUNET_YES;
 856   h->peers = GNUNET_CONTAINER_multipeermap_create (128, GNUNET_NO);
 857   hcnt = 0;
 858   if (NULL != handlers)
 859     while (NULL != handlers[hcnt].callback)
 860       hcnt++;
 861   h->handlers = GNUNET_new_array (hcnt + 1,
 862                                   struct GNUNET_CORE_MessageHandler);
 863   if (NULL != handlers)
 864     memcpy (h->handlers,
 865             handlers,
 866             hcnt * sizeof (struct GNUNET_CORE_MessageHandler));
 867   h->hcnt = hcnt;
 868   GNUNET_assert (hcnt <
 869                  (GNUNET_SERVER_MAX_MESSAGE_SIZE -
 870                   sizeof (struct InitMessage)) / sizeof (uint16_t));
 871   LOG (GNUNET_ERROR_TYPE_DEBUG,
 872        "Connecting to CORE service\n");
 873   reconnect (h);
 874   return h;
 875 }
 876
 877
 878 /**
 879  * Disconnect from the core service.  This function can only
 880  * be called *after* all pending #GNUNET_CORE_notify_transmit_ready()
 881  * requests have been explicitly canceled.
 882  *
 883  * @param handle connection to core to disconnect
 884  */
 885 void
 886 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
 887 {
 888   LOG (GNUNET_ERROR_TYPE_DEBUG,
 889        "Disconnecting from CORE service\n");
 890   GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
 891                                          &disconnect_and_free_peer_entry,
 892                                          handle);
 893   GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
 894   handle->peers = NULL;
 895   if (NULL != handle->reconnect_task)
 896   {
 897     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 898     handle->reconnect_task = NULL;
 899   }
 900   if (NULL != handle->mq)
 901   {
 902     GNUNET_MQ_destroy (handle->mq);
 903     handle->mq = NULL;
 904   }
 905   GNUNET_free (handle->handlers);
 906   GNUNET_free (handle);
 907 }
 908
 909
 910 /**
 911  * Ask the core to call @a notify once it is ready to transmit the
 912  * given number of bytes to the specified @a target.  Must only be
 913  * called after a connection to the respective peer has been
 914  * established (and the client has been informed about this).  You may
 915  * have one request of this type pending for each connected peer at
 916  * any time.  If a peer disconnects, the application MUST call
 917  * #GNUNET_CORE_notify_transmit_ready_cancel on the respective
 918  * transmission request, if one such request is pending.
 919  *
 920  * @param handle connection to core service
 921  * @param cork is corking allowed for this transmission?
 922  * @param priority how important is the message?
 923  * @param maxdelay how long can the message wait? Only effective if @a cork is #GNUNET_YES
 924  * @param target who should receive the message, never NULL (can be this peer's identity for loopback)
 925  * @param notify_size how many bytes of buffer space does @a notify want?
 926  * @param notify function to call when buffer space is available;
 927  *        will be called with NULL on timeout; clients MUST cancel
 928  *        all pending transmission requests DURING the disconnect
 929  *        handler
 930  * @param notify_cls closure for @a notify
 931  * @return non-NULL if the notify callback was queued,
 932  *         NULL if we can not even queue the request (request already pending);
 933  *         if NULL is returned, @a notify will NOT be called.
 934  */
 935 struct GNUNET_CORE_TransmitHandle *
 936 GNUNET_CORE_notify_transmit_ready (struct GNUNET_CORE_Handle *handle,
 937                                    int cork,
 938                                    enum GNUNET_CORE_Priority priority,
 939                                    struct GNUNET_TIME_Relative maxdelay,
 940                                    const struct GNUNET_PeerIdentity *target,
 941                                    size_t notify_size,
 942                                    GNUNET_CONNECTION_TransmitReadyNotify notify,
 943                                    void *notify_cls)
 944 {
 945   struct PeerRecord *pr;
 946   struct GNUNET_CORE_TransmitHandle *th;
 947   struct SendMessageRequest *smr;
 948   struct GNUNET_MQ_Envelope *env;
 949
 950   GNUNET_assert (NULL != notify);
 951   if ( (notify_size > GNUNET_CONSTANTS_MAX_ENCRYPTED_MESSAGE_SIZE) ||
 952        (notify_size + sizeof (struct SendMessage) >= GNUNET_SERVER_MAX_MESSAGE_SIZE) )
 953   {
 954     GNUNET_break (0);
 955     return NULL;
 956   }
 957   LOG (GNUNET_ERROR_TYPE_DEBUG,
 958        "Asking core for transmission of %u bytes to `%s'\n",
 959        (unsigned int) notify_size,
 960        GNUNET_i2s (target));
 961   pr = GNUNET_CONTAINER_multipeermap_get (handle->peers,
 962                                           target);
 963   if (NULL == pr)
 964   {
 965     /* attempt to send to peer that is not connected */
 966     GNUNET_break (0);
 967     return NULL;
 968   }
 969   if (NULL != pr->th.peer)
 970   {
 971     /* attempting to queue a second request for the same destination */
 972     GNUNET_break (0);
 973     return NULL;
 974   }
 975   th = &pr->th;
 976   memset (th,
 977           0,
 978           sizeof (struct GNUNET_CORE_TransmitHandle));
 979   th->peer = pr;
 980   th->get_message = notify;
 981   th->get_message_cls = notify_cls;
 982   th->request_time = GNUNET_TIME_absolute_get ();
 983   if (GNUNET_YES == cork)
 984     th->deadline = GNUNET_TIME_relative_to_absolute (maxdelay);
 985   else
 986     th->deadline = th->request_time;
 987   th->priority = priority;
 988   th->msize = notify_size;
 989   th->cork = cork;
 990   env = GNUNET_MQ_msg (smr,
 991                        GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
 992   smr->priority = htonl ((uint32_t) th->priority);
 993   smr->deadline = GNUNET_TIME_absolute_hton (th->deadline);
 994   smr->peer = pr->peer;
 995   smr->reserved = htonl (0);
 996   smr->size = htons (th->msize);
 997   smr->smr_id = htons (th->smr_id = pr->smr_id_gen++);
 998   GNUNET_MQ_send (handle->mq,
 999                   env);
1000   LOG (GNUNET_ERROR_TYPE_DEBUG,
1001        "Transmission request added to queue\n");
1002   return th;
1003 }
1004
1005
1006 /**
1007  * Cancel the specified transmission-ready notification.
1008  *
1009  * @param th handle that was returned by #GNUNET_CORE_notify_transmit_ready().
1010  */
1011 void
1012 GNUNET_CORE_notify_transmit_ready_cancel (struct GNUNET_CORE_TransmitHandle *th)
1013 {
1014   struct PeerRecord *pr = th->peer;
1015
1016   LOG (GNUNET_ERROR_TYPE_DEBUG,
1017        "Aborting transmission request to core for %u bytes to `%s'\n",
1018        (unsigned int) th->msize,
1019        GNUNET_i2s (&pr->peer));
1020   th->peer = NULL;
1021 }
1022
1023
1024 /**
1025  * Check if the given peer is currently connected. This function is for special
1026  * cirumstances (GNUNET_TESTBED uses it), normal users of the CORE API are
1027  * expected to track which peers are connected based on the connect/disconnect
1028  * callbacks from #GNUNET_CORE_connect().  This function is NOT part of the
1029  * 'versioned', 'official' API. The difference between this function and the
1030  * function GNUNET_CORE_is_peer_connected() is that this one returns
1031  * synchronously after looking in the CORE API cache. The function
1032  * GNUNET_CORE_is_peer_connected() sends a message to the CORE service and hence
1033  * its response is given asynchronously.
1034  *
1035  * @param h the core handle
1036  * @param pid the identity of the peer to check if it has been connected to us
1037  * @return #GNUNET_YES if the peer is connected to us; #GNUNET_NO if not
1038  */
1039 int
1040 GNUNET_CORE_is_peer_connected_sync (const struct GNUNET_CORE_Handle *h,
1041                                     const struct GNUNET_PeerIdentity *pid)
1042 {
1043   return GNUNET_CONTAINER_multipeermap_contains (h->peers,
1044                                                  pid);
1045 }
1046
1047
1048 /* end of core_api.c */