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 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 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  * Information we track for each peer.
  37  */
  38 struct PeerRecord
  39 {
  40
  41   /**
  42    * Corresponding CORE handle.
  43    */
  44   struct GNUNET_CORE_Handle *h;
  45
  46   /**
  47    * Message queue for the peer.
  48    */
  49   struct GNUNET_MQ_Handle *mq;
  50
  51   /**
  52    * Message we are currently trying to pass to the CORE service
  53    * for this peer (from @e mq).
  54    */
  55   struct GNUNET_MQ_Envelope *env;
  56
  57   /**
  58    * Value the client returned when we connected, used
  59    * as the closure in various places.
  60    */
  61   void *client_cls;
  62
  63   /**
  64    * Peer the record is about.
  65    */
  66   struct GNUNET_PeerIdentity peer;
  67
  68   /**
  69    * SendMessageRequest ID generator for this peer.
  70    */
  71   uint16_t smr_id_gen;
  72
  73 };
  74
  75
  76 /**
  77  * Context for the core service connection.
  78  */
  79 struct GNUNET_CORE_Handle
  80 {
  81
  82   /**
  83    * Configuration we're using.
  84    */
  85   const struct GNUNET_CONFIGURATION_Handle *cfg;
  86
  87   /**
  88    * Closure for the various callbacks.
  89    */
  90   void *cls;
  91
  92   /**
  93    * Function to call once we've handshaked with the core service.
  94    */
  95   GNUNET_CORE_StartupCallback init;
  96
  97   /**
  98    * Function to call whenever we're notified about a peer connecting.
  99    */
 100   GNUNET_CORE_ConnectEventHandler connects;
 101
 102   /**
 103    * Function to call whenever we're notified about a peer disconnecting.
 104    */
 105   GNUNET_CORE_DisconnectEventHandler disconnects;
 106
 107   /**
 108    * Function handlers for messages of particular type.
 109    */
 110   struct GNUNET_MQ_MessageHandler *handlers;
 111
 112   /**
 113    * Our message queue for transmissions to the service.
 114    */
 115   struct GNUNET_MQ_Handle *mq;
 116
 117   /**
 118    * Hash map listing all of the peers that we are currently
 119    * connected to.
 120    */
 121   struct GNUNET_CONTAINER_MultiPeerMap *peers;
 122
 123   /**
 124    * Identity of this peer.
 125    */
 126   struct GNUNET_PeerIdentity me;
 127
 128   /**
 129    * ID of reconnect task (if any).
 130    */
 131   struct GNUNET_SCHEDULER_Task *reconnect_task;
 132
 133   /**
 134    * Current delay we use for re-trying to connect to core.
 135    */
 136   struct GNUNET_TIME_Relative retry_backoff;
 137
 138   /**
 139    * Number of entries in the handlers array.
 140    */
 141   unsigned int hcnt;
 142
 143   /**
 144    * Did we ever get INIT?
 145    */
 146   int have_init;
 147
 148 };
 149
 150
 151 /**
 152  * Our current client connection went down.  Clean it up
 153  * and try to reconnect!
 154  *
 155  * @param h our handle to the core service
 156  */
 157 static void
 158 reconnect (struct GNUNET_CORE_Handle *h);
 159
 160
 161 /**
 162  * Task schedule to try to re-connect to core.
 163  *
 164  * @param cls the `struct GNUNET_CORE_Handle`
 165  * @param tc task context
 166  */
 167 static void
 168 reconnect_task (void *cls)
 169 {
 170   struct GNUNET_CORE_Handle *h = cls;
 171
 172   h->reconnect_task = NULL;
 173   LOG (GNUNET_ERROR_TYPE_DEBUG,
 174        "Connecting to CORE service after delay\n");
 175   reconnect (h);
 176 }
 177
 178
 179 /**
 180  * Notify clients about disconnect and free the entry for connected
 181  * peer.
 182  *
 183  * @param cls the `struct GNUNET_CORE_Handle *`
 184  * @param key the peer identity (not used)
 185  * @param value the `struct PeerRecord` to free.
 186  * @return #GNUNET_YES (continue)
 187  */
 188 static int
 189 disconnect_and_free_peer_entry (void *cls,
 190                                 const struct GNUNET_PeerIdentity *key,
 191                                 void *value)
 192 {
 193   struct GNUNET_CORE_Handle *h = cls;
 194   struct PeerRecord *pr = value;
 195
 196   GNUNET_assert (pr->h == h);
 197   if (NULL != h->disconnects)
 198     h->disconnects (h->cls,
 199                     &pr->peer,
 200                     pr->client_cls);
 201   GNUNET_assert (GNUNET_YES ==
 202                  GNUNET_CONTAINER_multipeermap_remove (h->peers,
 203                                                        key,
 204                                                        pr));
 205   GNUNET_MQ_destroy (pr->mq);
 206   GNUNET_assert (NULL == pr->mq);
 207   if (NULL != pr->env)
 208   {
 209     GNUNET_MQ_discard (pr->env);
 210     pr->env = NULL;
 211   }
 212   GNUNET_free (pr);
 213   return GNUNET_YES;
 214 }
 215
 216
 217 /**
 218  * Close down any existing connection to the CORE service and
 219  * try re-establishing it later.
 220  *
 221  * @param h our handle
 222  */
 223 static void
 224 reconnect_later (struct GNUNET_CORE_Handle *h)
 225 {
 226   GNUNET_assert (NULL == h->reconnect_task);
 227   if (NULL != h->mq)
 228   {
 229     GNUNET_MQ_destroy (h->mq);
 230     h->mq = NULL;
 231   }
 232   GNUNET_assert (NULL == h->reconnect_task);
 233   h->reconnect_task =
 234       GNUNET_SCHEDULER_add_delayed (h->retry_backoff,
 235                                     &reconnect_task,
 236                                     h);
 237   GNUNET_CONTAINER_multipeermap_iterate (h->peers,
 238                                          &disconnect_and_free_peer_entry,
 239                                          h);
 240   h->retry_backoff = GNUNET_TIME_STD_BACKOFF (h->retry_backoff);
 241 }
 242
 243
 244 /**
 245  * Error handler for the message queue to the CORE service.
 246  * On errors, we reconnect.
 247  *
 248  * @param cls closure, a `struct GNUNET_CORE_Handle *`
 249  * @param error error code
 250  */
 251 static void
 252 handle_mq_error (void *cls,
 253                  enum GNUNET_MQ_Error error)
 254 {
 255   struct GNUNET_CORE_Handle *h = cls;
 256
 257   LOG (GNUNET_ERROR_TYPE_DEBUG,
 258        "MQ ERROR: %d\n",
 259        error);
 260   reconnect_later (h);
 261 }
 262
 263
 264 /**
 265  * Inquire with CORE what options should be set for a message
 266  * so that it is transmitted with the given @a priority and
 267  * the given @a cork value.
 268  *
 269  * @param cork desired corking
 270  * @param priority desired message priority
 271  * @param[out] flags set to `flags` value for #GNUNET_MQ_set_options()
 272  * @return `extra` argument to give to #GNUNET_MQ_set_options()
 273  */
 274 const void *
 275 GNUNET_CORE_get_mq_options (int cork,
 276                             enum GNUNET_CORE_Priority priority,
 277                             uint64_t *flags)
 278 {
 279   *flags = ((uint64_t) priority) + (((uint64_t) cork) << 32);
 280   return NULL;
 281 }
 282
 283
 284 /**
 285  * Implement sending functionality of a message queue for
 286  * us sending messages to a peer.
 287  *
 288  * @param mq the message queue
 289  * @param msg the message to send
 290  * @param impl_state state of the implementation
 291  */
 292 static void
 293 core_mq_send_impl (struct GNUNET_MQ_Handle *mq,
 294                    const struct GNUNET_MessageHeader *msg,
 295                    void *impl_state)
 296 {
 297   struct PeerRecord *pr = impl_state;
 298   struct GNUNET_CORE_Handle *h = pr->h;
 299   struct SendMessageRequest *smr;
 300   struct SendMessage *sm;
 301   struct GNUNET_MQ_Envelope *env;
 302   uint16_t msize;
 303   uint64_t flags;
 304   int cork;
 305   enum GNUNET_CORE_Priority priority;
 306
 307   if (NULL == h->mq)
 308   {
 309     /* We're currently reconnecting, pretend this worked */
 310     GNUNET_MQ_impl_send_continue (mq);
 311     return;
 312   }
 313   GNUNET_assert (NULL == pr->env);
 314   /* extract options from envelope */
 315   env = GNUNET_MQ_get_current_envelope (mq);
 316   GNUNET_break (NULL ==
 317                 GNUNET_MQ_env_get_options (env,
 318                                            &flags));
 319   cork = (int) (flags >> 32);
 320   priority = (uint32_t) flags;
 321
 322   /* check message size for sanity */
 323   msize = ntohs (msg->size);
 324   if (msize >= GNUNET_MAX_MESSAGE_SIZE - sizeof (struct SendMessage))
 325   {
 326     GNUNET_break (0);
 327     GNUNET_MQ_impl_send_continue (mq);
 328     return;
 329   }
 330
 331   /* ask core for transmission */
 332   LOG (GNUNET_ERROR_TYPE_DEBUG,
 333        "Asking core for transmission of %u bytes to `%s'\n",
 334        (unsigned int) msize,
 335        GNUNET_i2s (&pr->peer));
 336   env = GNUNET_MQ_msg (smr,
 337                        GNUNET_MESSAGE_TYPE_CORE_SEND_REQUEST);
 338   smr->priority = htonl ((uint32_t) priority);
 339   smr->peer = pr->peer;
 340   smr->reserved = htonl (0);
 341   smr->size = htons (msize);
 342   smr->smr_id = htons (++pr->smr_id_gen);
 343   GNUNET_MQ_send (h->mq,
 344                   env);
 345
 346   /* prepare message with actual transmission data */
 347   pr->env = GNUNET_MQ_msg_nested_mh (sm,
 348                                      GNUNET_MESSAGE_TYPE_CORE_SEND,
 349                                      msg);
 350   sm->priority = htonl ((uint32_t) priority);
 351   sm->peer = pr->peer;
 352   sm->cork = htonl ((uint32_t) cork);
 353   sm->reserved = htonl (0);
 354   LOG (GNUNET_ERROR_TYPE_DEBUG,
 355        "Calling get_message with buffer of %u bytes (%s)\n",
 356        (unsigned int) msize,
 357        cork ? "corked" : "uncorked");
 358 }
 359
 360
 361 /**
 362  * Handle destruction of a message queue.  Implementations must not
 363  * free @a mq, but should take care of @a impl_state.
 364  *
 365  * @param mq the message queue to destroy
 366  * @param impl_state state of the implementation
 367  */
 368 static void
 369 core_mq_destroy_impl (struct GNUNET_MQ_Handle *mq,
 370                       void *impl_state)
 371 {
 372   struct PeerRecord *pr = impl_state;
 373
 374   GNUNET_assert (mq == pr->mq);
 375   pr->mq = NULL;
 376 }
 377
 378
 379 /**
 380  * Implementation function that cancels the currently sent message.
 381  * Should basically undo whatever #mq_send_impl() did.
 382  *
 383  * @param mq message queue
 384  * @param impl_state state specific to the implementation
 385  */
 386 static void
 387 core_mq_cancel_impl (struct GNUNET_MQ_Handle *mq,
 388                      void *impl_state)
 389 {
 390   struct PeerRecord *pr = impl_state;
 391
 392   GNUNET_assert (NULL != pr->env);
 393   GNUNET_MQ_discard (pr->env);
 394   pr->env = NULL;
 395 }
 396
 397
 398 /**
 399  * We had an error processing a message we forwarded from a peer to
 400  * the CORE service.  We should just complain about it but otherwise
 401  * continue processing.
 402  *
 403  * @param cls closure
 404  * @param error error code
 405  */
 406 static void
 407 core_mq_error_handler (void *cls,
 408                        enum GNUNET_MQ_Error error)
 409 {
 410   /* struct PeerRecord *pr = cls; */
 411
 412   GNUNET_break_op (0);
 413 }
 414
 415
 416 /**
 417  * Add the given peer to the list of our connected peers
 418  * and create the respective data structures and notify
 419  * the application.
 420  *
 421  * @param h the core handle
 422  * @param peer the peer that is connecting to us
 423  */
 424 static void
 425 connect_peer (struct GNUNET_CORE_Handle *h,
 426               const struct GNUNET_PeerIdentity *peer)
 427 {
 428   struct PeerRecord *pr;
 429   uint64_t flags;
 430   const void *extra;
 431
 432   pr = GNUNET_new (struct PeerRecord);
 433   pr->peer = *peer;
 434   pr->h = h;
 435   GNUNET_assert (GNUNET_YES ==
 436                  GNUNET_CONTAINER_multipeermap_put (h->peers,
 437                                                     &pr->peer,
 438                                                     pr,
 439                                                     GNUNET_CONTAINER_MULTIHASHMAPOPTION_UNIQUE_ONLY));
 440   pr->mq = GNUNET_MQ_queue_for_callbacks (&core_mq_send_impl,
 441                                           &core_mq_destroy_impl,
 442                                           &core_mq_cancel_impl,
 443                                           pr,
 444                                           h->handlers,
 445                                           &core_mq_error_handler,
 446                                           pr);
 447   /* get our default options */
 448   extra = GNUNET_CORE_get_mq_options (GNUNET_NO,
 449                                       GNUNET_CORE_PRIO_BEST_EFFORT,
 450                                       &flags);
 451   GNUNET_MQ_set_options (pr->mq,
 452                          flags,
 453                          extra);
 454   if (NULL != h->connects)
 455   {
 456     pr->client_cls = h->connects (h->cls,
 457                                   &pr->peer,
 458                                   pr->mq);
 459     GNUNET_MQ_set_handlers_closure (pr->mq,
 460                                     pr->client_cls);
 461   }
 462 }
 463
 464
 465 /**
 466  * Handle  init  reply message  received  from  CORE service.   Notify
 467  * application  that we  are now  connected  to the  CORE.  Also  fake
 468  * loopback connection.
 469  *
 470  * @param cls the `struct GNUNET_CORE_Handle`
 471  * @param m the init reply
 472  */
 473 static void
 474 handle_init_reply (void *cls,
 475                    const struct InitReplyMessage *m)
 476 {
 477   struct GNUNET_CORE_Handle *h = cls;
 478   GNUNET_CORE_StartupCallback init;
 479
 480   GNUNET_break (0 == ntohl (m->reserved));
 481   h->retry_backoff = GNUNET_TIME_UNIT_MILLISECONDS;
 482   if (NULL != (init = h->init))
 483   {
 484     /* mark so we don't call init on reconnect */
 485     h->init = NULL;
 486     h->me = m->my_identity;
 487     LOG (GNUNET_ERROR_TYPE_DEBUG,
 488          "Connected to core service of peer `%s'.\n",
 489          GNUNET_i2s (&h->me));
 490     h->have_init = GNUNET_YES;
 491     init (h->cls,
 492           &h->me);
 493   }
 494   else
 495   {
 496     LOG (GNUNET_ERROR_TYPE_DEBUG,
 497          "Successfully reconnected to core service.\n");
 498     if (GNUNET_NO == h->have_init)
 499     {
 500       h->me = m->my_identity;
 501       h->have_init = GNUNET_YES;
 502     }
 503     else
 504     {
 505       GNUNET_break (0 == memcmp (&h->me,
 506                                  &m->my_identity,
 507                                  sizeof (struct GNUNET_PeerIdentity)));
 508     }
 509   }
 510   /* fake 'connect to self' */
 511   connect_peer (h,
 512                 &h->me);
 513 }
 514
 515
 516 /**
 517  * Handle connect message received from CORE service.
 518  * Notify the application about the new connection.
 519  *
 520  * @param cls the `struct GNUNET_CORE_Handle`
 521  * @param cnm the connect message
 522  */
 523 static void
 524 handle_connect_notify (void *cls,
 525                        const struct ConnectNotifyMessage *cnm)
 526 {
 527   struct GNUNET_CORE_Handle *h = cls;
 528   struct PeerRecord *pr;
 529
 530   LOG (GNUNET_ERROR_TYPE_DEBUG,
 531        "Received notification about connection from `%s'.\n",
 532        GNUNET_i2s (&cnm->peer));
 533   if (0 == memcmp (&h->me,
 534                    &cnm->peer,
 535                    sizeof (struct GNUNET_PeerIdentity)))
 536   {
 537     /* connect to self!? */
 538     GNUNET_break (0);
 539     return;
 540   }
 541   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 542                                           &cnm->peer);
 543   if (NULL != pr)
 544   {
 545     GNUNET_break (0);
 546     reconnect_later (h);
 547     return;
 548   }
 549   connect_peer (h,
 550                 &cnm->peer);
 551 }
 552
 553
 554 /**
 555  * Handle disconnect message received from CORE service.
 556  * Notify the application about the lost connection.
 557  *
 558  * @param cls the `struct GNUNET_CORE_Handle`
 559  * @param dnm message about the disconnect event
 560  */
 561 static void
 562 handle_disconnect_notify (void *cls,
 563                           const struct DisconnectNotifyMessage *dnm)
 564 {
 565   struct GNUNET_CORE_Handle *h = cls;
 566   struct PeerRecord *pr;
 567
 568   if (0 == memcmp (&h->me,
 569                    &dnm->peer,
 570                    sizeof (struct GNUNET_PeerIdentity)))
 571   {
 572     /* disconnect from self!? */
 573     GNUNET_break (0);
 574     return;
 575   }
 576   GNUNET_break (0 == ntohl (dnm->reserved));
 577   LOG (GNUNET_ERROR_TYPE_DEBUG,
 578        "Received notification about disconnect from `%s'.\n",
 579        GNUNET_i2s (&dnm->peer));
 580   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 581                                           &dnm->peer);
 582   if (NULL == pr)
 583   {
 584     GNUNET_break (0);
 585     reconnect_later (h);
 586     return;
 587   }
 588   disconnect_and_free_peer_entry (h,
 589                                   &pr->peer,
 590                                   pr);
 591 }
 592
 593
 594 /**
 595  * Check that message received from CORE service is well-formed.
 596  *
 597  * @param cls the `struct GNUNET_CORE_Handle`
 598  * @param ntm the message we got
 599  * @return #GNUNET_OK if the message is well-formed
 600  */
 601 static int
 602 check_notify_inbound (void *cls,
 603                       const struct NotifyTrafficMessage *ntm)
 604 {
 605   uint16_t msize;
 606   const struct GNUNET_MessageHeader *em;
 607
 608   msize = ntohs (ntm->header.size) - sizeof (struct NotifyTrafficMessage);
 609   if (msize < sizeof (struct GNUNET_MessageHeader))
 610   {
 611     GNUNET_break (0);
 612     return GNUNET_SYSERR;
 613   }
 614   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 615   if (msize != ntohs (em->size))
 616   {
 617     GNUNET_break (0);
 618     return GNUNET_SYSERR;
 619   }
 620   return GNUNET_OK;
 621 }
 622
 623
 624 /**
 625  * Handle inbound message received from CORE service.  If applicable,
 626  * notify the application.
 627  *
 628  * @param cls the `struct GNUNET_CORE_Handle`
 629  * @param ntm the message we got from CORE.
 630  */
 631 static void
 632 handle_notify_inbound (void *cls,
 633                        const struct NotifyTrafficMessage *ntm)
 634 {
 635   struct GNUNET_CORE_Handle *h = cls;
 636   const struct GNUNET_MessageHeader *em;
 637   struct PeerRecord *pr;
 638
 639   LOG (GNUNET_ERROR_TYPE_DEBUG,
 640        "Received inbound message from `%s'.\n",
 641        GNUNET_i2s (&ntm->peer));
 642   em = (const struct GNUNET_MessageHeader *) &ntm[1];
 643   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 644                                           &ntm->peer);
 645   if (NULL == pr)
 646   {
 647     GNUNET_break (0);
 648     reconnect_later (h);
 649     return;
 650   }
 651   GNUNET_MQ_inject_message (pr->mq,
 652                             em);
 653 }
 654
 655
 656 /**
 657  * Handle message received from CORE service notifying us that we are
 658  * now allowed to send a message to a peer.  If that message is still
 659  * pending, put it into the queue to be transmitted.
 660  *
 661  * @param cls the `struct GNUNET_CORE_Handle`
 662  * @param smr the message we got
 663  */
 664 static void
 665 handle_send_ready (void *cls,
 666                    const struct SendMessageReady *smr)
 667 {
 668   struct GNUNET_CORE_Handle *h = cls;
 669   struct PeerRecord *pr;
 670
 671   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 672                                           &smr->peer);
 673   if (NULL == pr)
 674   {
 675     GNUNET_break (0);
 676     reconnect_later (h);
 677     return;
 678   }
 679   LOG (GNUNET_ERROR_TYPE_DEBUG,
 680        "Received notification about transmission readiness to `%s'.\n",
 681        GNUNET_i2s (&smr->peer));
 682   if (NULL == pr->env)
 683   {
 684     /* request must have been cancelled between the original request
 685      * and the response from CORE, ignore CORE's readiness */
 686     return;
 687   }
 688   if (ntohs (smr->smr_id) != pr->smr_id_gen)
 689   {
 690     /* READY message is for expired or cancelled message,
 691      * ignore! (we should have already sent another request) */
 692     return;
 693   }
 694
 695   /* ok, all good, send message out! */
 696   GNUNET_MQ_send (h->mq,
 697                   pr->env);
 698   pr->env = NULL;
 699   GNUNET_MQ_impl_send_continue (pr->mq);
 700 }
 701
 702
 703 /**
 704  * Our current client connection went down.  Clean it up and try to
 705  * reconnect!
 706  *
 707  * @param h our handle to the core service
 708  */
 709 static void
 710 reconnect (struct GNUNET_CORE_Handle *h)
 711 {
 712   struct GNUNET_MQ_MessageHandler handlers[] = {
 713     GNUNET_MQ_hd_fixed_size (init_reply,
 714                              GNUNET_MESSAGE_TYPE_CORE_INIT_REPLY,
 715                              struct InitReplyMessage,
 716                              h),
 717     GNUNET_MQ_hd_fixed_size (connect_notify,
 718                              GNUNET_MESSAGE_TYPE_CORE_NOTIFY_CONNECT,
 719                              struct ConnectNotifyMessage,
 720                              h),
 721     GNUNET_MQ_hd_fixed_size (disconnect_notify,
 722                              GNUNET_MESSAGE_TYPE_CORE_NOTIFY_DISCONNECT,
 723                              struct DisconnectNotifyMessage,
 724                              h),
 725     GNUNET_MQ_hd_var_size (notify_inbound,
 726                            GNUNET_MESSAGE_TYPE_CORE_NOTIFY_INBOUND,
 727                            struct NotifyTrafficMessage,
 728                            h),
 729     GNUNET_MQ_hd_fixed_size (send_ready,
 730                              GNUNET_MESSAGE_TYPE_CORE_SEND_READY,
 731                              struct SendMessageReady,
 732                              h),
 733     GNUNET_MQ_handler_end ()
 734   };
 735   struct InitMessage *init;
 736   struct GNUNET_MQ_Envelope *env;
 737   uint16_t *ts;
 738
 739   GNUNET_assert (NULL == h->mq);
 740   h->mq = GNUNET_CLIENT_connect (h->cfg,
 741                                  "core",
 742                                  handlers,
 743                                  &handle_mq_error,
 744                                  h);
 745   if (NULL == h->mq)
 746   {
 747     reconnect_later (h);
 748     return;
 749   }
 750   env = GNUNET_MQ_msg_extra (init,
 751                              sizeof (uint16_t) * h->hcnt,
 752                              GNUNET_MESSAGE_TYPE_CORE_INIT);
 753   LOG (GNUNET_ERROR_TYPE_INFO,
 754        "(Re)connecting to CORE service\n");
 755   init->options = htonl (0);
 756   ts = (uint16_t *) &init[1];
 757   for (unsigned int hpos = 0; hpos < h->hcnt; hpos++)
 758     ts[hpos] = htons (h->handlers[hpos].type);
 759   GNUNET_MQ_send (h->mq,
 760                   env);
 761 }
 762
 763
 764 /**
 765  * Connect to the core service.  Note that the connection may complete
 766  * (or fail) asynchronously.
 767  *
 768  * @param cfg configuration to use
 769  * @param cls closure for the various callbacks that follow (including handlers in the handlers array)
 770  * @param init callback to call once we have successfully
 771  *        connected to the core service
 772  * @param connects function to call on peer connect, can be NULL
 773  * @param disconnects function to call on peer disconnect / timeout, can be NULL
 774  * @param handlers callbacks for messages we care about, NULL-terminated
 775  * @return handle to the core service (only useful for disconnect until @a init is called);
 776  *                NULL on error (in this case, init is never called)
 777  */
 778 struct GNUNET_CORE_Handle *
 779 GNUNET_CORE_connect (const struct GNUNET_CONFIGURATION_Handle *cfg,
 780                      void *cls,
 781                      GNUNET_CORE_StartupCallback init,
 782                      GNUNET_CORE_ConnectEventHandler connects,
 783                      GNUNET_CORE_DisconnectEventHandler disconnects,
 784                      const struct GNUNET_MQ_MessageHandler *handlers)
 785 {
 786   struct GNUNET_CORE_Handle *h;
 787
 788   h = GNUNET_new (struct GNUNET_CORE_Handle);
 789   h->cfg = cfg;
 790   h->cls = cls;
 791   h->init = init;
 792   h->connects = connects;
 793   h->disconnects = disconnects;
 794   h->peers = GNUNET_CONTAINER_multipeermap_create (128,
 795                                                    GNUNET_NO);
 796   h->handlers = GNUNET_MQ_copy_handlers (handlers);
 797   h->hcnt = GNUNET_MQ_count_handlers (handlers);
 798   GNUNET_assert (h->hcnt <
 799                  (GNUNET_MAX_MESSAGE_SIZE -
 800                   sizeof (struct InitMessage)) / sizeof (uint16_t));
 801   LOG (GNUNET_ERROR_TYPE_DEBUG,
 802        "Connecting to CORE service\n");
 803   reconnect (h);
 804   if (NULL == h->mq)
 805   {
 806     GNUNET_CORE_disconnect (h);
 807     return NULL;
 808   }
 809   return h;
 810 }
 811
 812
 813 /**
 814  * Disconnect from the core service.
 815  *
 816  * @param handle connection to core to disconnect
 817  */
 818 void
 819 GNUNET_CORE_disconnect (struct GNUNET_CORE_Handle *handle)
 820 {
 821   LOG (GNUNET_ERROR_TYPE_DEBUG,
 822        "Disconnecting from CORE service\n");
 823   GNUNET_CONTAINER_multipeermap_iterate (handle->peers,
 824                                          &disconnect_and_free_peer_entry,
 825                                          handle);
 826   GNUNET_CONTAINER_multipeermap_destroy (handle->peers);
 827   handle->peers = NULL;
 828   if (NULL != handle->reconnect_task)
 829   {
 830     GNUNET_SCHEDULER_cancel (handle->reconnect_task);
 831     handle->reconnect_task = NULL;
 832   }
 833   if (NULL != handle->mq)
 834   {
 835     GNUNET_MQ_destroy (handle->mq);
 836     handle->mq = NULL;
 837   }
 838   GNUNET_free_non_null (handle->handlers);
 839   GNUNET_free (handle);
 840 }
 841
 842
 843 /**
 844  * Obtain the message queue for a connected peer.
 845  *
 846  * @param h the core handle
 847  * @param pid the identity of the peer to check if it has been connected to us
 848  * @return NULL if peer is not connected
 849  */
 850 struct GNUNET_MQ_Handle *
 851 GNUNET_CORE_get_mq (const struct GNUNET_CORE_Handle *h,
 852                     const struct GNUNET_PeerIdentity *pid)
 853 {
 854   struct PeerRecord *pr;
 855
 856   pr = GNUNET_CONTAINER_multipeermap_get (h->peers,
 857                                           pid);
 858   if (NULL == pr)
 859     return NULL;
 860   return pr->mq;
 861 }
 862
 863
 864 /* end of core_api.c */