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