src/transport/transport-testing2.c

   1 /*
   2      This file is part of GNUnet.
   3      Copyright (C) 2019 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 /**
  22  * @file transport/transport-testing2.c
  23  * @brief functions related to testing-tng
  24  * @author Christian Grothoff
  25  * @author Julius Bünger
  26  */
  27 #include "platform.h"
  28 #include "gnunet_util_lib.h"
  29 #include "gnunet_protocols.h"
  30 #include "gnunet_constants.h"
  31 #include "transport-testing2.h"
  32 #include "gnunet_ats_transport_service.h"
  33 #include "gnunet_hello_lib.h"
  34 #include "gnunet_signatures.h"
  35 #include "transport.h"
  36
  37
  38 #define LOG(kind, ...) GNUNET_log_from (kind, "transport-testing2", __VA_ARGS__)
  39
  40
  41 /**
  42  * @brief Handle to a transport communicator
  43  */
  44 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle
  45 {
  46   /**
  47    * @brief Handle to the configuration
  48    */
  49   struct GNUNET_CONFIGURATION_Handle *cfg;
  50
  51   /**
  52    * @brief File name of configuration file
  53    */
  54   char *cfg_filename;
  55
  56   /**
  57    * @brief Handle to the transport service
  58    */
  59   struct GNUNET_SERVICE_Handle *tsh;
  60
  61   /**
  62    * @brief Task that will be run on shutdown to stop and clean transport
  63    * service
  64    */
  65   struct GNUNET_SCHEDULER_Task *ts_shutdown_task;
  66
  67   /**
  68    * @brief Handle to the client
  69    */
  70   struct GNUNET_SERVICE_Client *client;
  71
  72   /**
  73    * @brief Handle to the client
  74    */
  75   struct GNUNET_MQ_Handle *c_mq;
  76
  77   /**
  78    * @brief Process of the communicator
  79    */
  80   struct GNUNET_OS_Process *c_proc;
  81
  82   /**
  83    * @brief Task that will be run on shutdown to stop and clean communicator
  84    */
  85   struct GNUNET_SCHEDULER_Task *c_shutdown_task;
  86
  87   /**
  88    * @brief Characteristics of the communicator
  89    */
  90   enum GNUNET_TRANSPORT_CommunicatorCharacteristics c_characteristics;
  91
  92   /**
  93    * @brief Specifies supported addresses
  94    */
  95   char *c_addr_prefix;
  96
  97   /**
  98    * @brief Specifies supported addresses
  99    */
 100   char *c_address;
 101
 102   /**
 103    * @brief Head of the DLL of queues associated with this communicator
 104    */
 105   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *queue_head;
 106
 107   /**
 108    * @brief Tail of the DLL of queues associated with this communicator
 109    */
 110   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *queue_tail;
 111
 112   /* Callbacks + Closures */
 113   /**
 114    * @brief Callback called when a new communicator connects
 115    */
 116   GNUNET_TRANSPORT_TESTING_CommunicatorAvailableCallback
 117     communicator_available_cb;
 118
 119   /**
 120    * @brief Callback called when a new communicator connects
 121    */
 122   GNUNET_TRANSPORT_TESTING_AddAddressCallback add_address_cb;
 123
 124   /**
 125    * @brief Callback called when a new communicator connects
 126    */
 127   GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback queue_create_reply_cb;
 128
 129   /**
 130    * @brief Callback called when a new communicator connects
 131    */
 132   GNUNET_TRANSPORT_TESTING_AddQueueCallback add_queue_cb;
 133
 134   /**
 135    * @brief Callback called when a new communicator connects
 136    */
 137   GNUNET_TRANSPORT_TESTING_IncomingMessageCallback incoming_msg_cb;
 138
 139   /**
 140    * @brief Closure to the callback
 141    */
 142   void *cb_cls;
 143 };
 144
 145
 146 /**
 147  * @brief Queue of a communicator and some context
 148  */
 149 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue
 150 {
 151   /**
 152    * @brief Handle to the TransportCommunicator
 153    */
 154   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h;
 155
 156   /**
 157    * @brief Envelope to a message that requests the opening of the queue.
 158    *
 159    * If the client already requests queue(s), but the communicator is not yet
 160    * connected, we cannot send the request to open the queue. Save it until the
 161    * communicator becomes available and send it then.
 162    */
 163   struct GNUNET_MQ_Envelope *open_queue_env;
 164
 165   /**
 166    * @brief Peer ID of the peer on the other side of the queue
 167    */
 168   struct GNUNET_PeerIdentity peer_id;
 169
 170   /**
 171    * @brief Queue ID
 172    */
 173   uint32_t qid;
 174
 175   /**
 176    * @brief Current message id
 177    */
 178   uint64_t mid;
 179
 180   /**
 181    * An `enum GNUNET_NetworkType` in NBO.
 182    */
 183   uint32_t nt;
 184
 185   /**
 186    * Maximum transmission unit, in NBO.  UINT32_MAX for unlimited.
 187    */
 188   uint32_t mtu;
 189
 190   /**
 191    * An `enum GNUNET_TRANSPORT_ConnectionStatus` in NBO.
 192    */
 193   uint32_t cs;
 194
 195   /**
 196    * @brief Next element inside a DLL
 197    */
 198   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *next;
 199
 200   /**
 201    * @brief Previous element inside a DLL
 202    */
 203   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *prev;
 204 };
 205
 206
 207 /**
 208  * @brief Handle/Context to a single transmission
 209  */
 210 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission
 211 {
 212 };
 213
 214
 215 /**
 216  * @brief Check whether incoming msg indicating available communicator is
 217  * correct
 218  *
 219  * @param cls Closure
 220  * @param msg Message struct
 221  *
 222  * @return GNUNET_YES in case message is correct
 223  */
 224 static int
 225 check_communicator_available (
 226   void *cls,
 227   const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *msg)
 228 {
 229   uint16_t size;
 230
 231   size = ntohs (msg->header.size) - sizeof(*msg);
 232   if (0 == size)
 233     return GNUNET_OK; /* receive-only communicator */
 234   GNUNET_MQ_check_zero_termination (msg);
 235   return GNUNET_OK;
 236 }
 237
 238
 239 /**
 240  * @brief Handle new communicator
 241  *
 242  * Store characteristics of communicator, call respective client callback.
 243  *
 244  * @param cls Closure - communicator handle
 245  * @param msg Message struct
 246  */
 247 static void
 248 handle_communicator_available (
 249   void *cls,
 250   const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *msg)
 251 {
 252   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 253   uint16_t size;
 254
 255   size = ntohs (msg->header.size) - sizeof(*msg);
 256   if (0 == size)
 257     return; /* receive-only communicator */
 258   tc_h->c_characteristics = ntohl (msg->cc);
 259   tc_h->c_addr_prefix = GNUNET_strdup ((const char *) &msg[1]);
 260   if (NULL != tc_h->communicator_available_cb)
 261   {
 262     LOG (GNUNET_ERROR_TYPE_DEBUG, "calling communicator_available_cb()\n");
 263     tc_h->communicator_available_cb (tc_h->cb_cls,
 264                                      tc_h,
 265                                      tc_h->c_characteristics,
 266                                      tc_h->c_addr_prefix);
 267   }
 268   GNUNET_SERVICE_client_continue (tc_h->client);
 269 }
 270
 271
 272 /**
 273  * Address of our peer added.  Test message is well-formed.
 274  *
 275  * @param cls the client
 276  * @param aam the send message that was sent
 277  * @return #GNUNET_OK if message is well-formed
 278  */
 279 static int
 280 check_add_address (void *cls,
 281                    const struct GNUNET_TRANSPORT_AddAddressMessage *msg)
 282 {
 283   struct TransportClient *tc = cls;
 284
 285   // if (CT_COMMUNICATOR != tc->type)
 286   // {
 287   //  GNUNET_break (0);
 288   //  return GNUNET_SYSERR;
 289   // }
 290   GNUNET_MQ_check_zero_termination (msg);
 291   return GNUNET_OK;
 292 }
 293
 294
 295 /**
 296  * @brief The communicator informs about an address.
 297  *
 298  * Store address and call client callback.
 299  *
 300  * @param cls Closure - communicator handle
 301  * @param msg Message
 302  */
 303 static void
 304 handle_add_address (void *cls,
 305                     const struct GNUNET_TRANSPORT_AddAddressMessage *msg)
 306 {
 307   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 308   uint16_t size;
 309   size = ntohs (msg->header.size) - sizeof(*msg);
 310   if (0 == size)
 311     return; /* receive-only communicator */
 312   LOG (GNUNET_ERROR_TYPE_DEBUG, "received add address cb %u\n", size);
 313   tc_h->c_address = GNUNET_strdup ((const char *) &msg[1]);
 314   if (NULL != tc_h->add_address_cb)
 315   {
 316     LOG (GNUNET_ERROR_TYPE_DEBUG, "calling add_address_cb()\n");
 317     tc_h->add_address_cb (tc_h->cb_cls,
 318                           tc_h,
 319                           tc_h->c_address,
 320                           GNUNET_TIME_relative_ntoh (msg->expiration),
 321                           msg->aid,
 322                           ntohl (msg->nt));
 323   }
 324   GNUNET_SERVICE_client_continue (tc_h->client);
 325 }
 326
 327
 328 /**
 329  * Incoming message.  Test message is well-formed.
 330  *
 331  * @param cls the client
 332  * @param msg the send message that was sent
 333  * @return #GNUNET_OK if message is well-formed
 334  */
 335 static int
 336 check_incoming_msg (void *cls,
 337                     const struct GNUNET_TRANSPORT_IncomingMessage *msg)
 338 {
 339   // struct TransportClient *tc = cls;
 340
 341   // if (CT_COMMUNICATOR != tc->type)
 342   // {
 343   //  GNUNET_break (0);
 344   //  return GNUNET_SYSERR;
 345   // }
 346   GNUNET_MQ_check_boxed_message (msg);
 347   return GNUNET_OK;
 348 }
 349
 350
 351 /**
 352  * @brief Receive an incoming message.
 353  *
 354  * Pass the message to the client.
 355  *
 356  * @param cls Closure - communicator handle
 357  * @param msg Message
 358  */
 359 static void
 360 handle_incoming_msg (void *cls,
 361                      const struct GNUNET_TRANSPORT_IncomingMessage *msg)
 362 {
 363   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 364
 365   if (NULL != tc_h->incoming_msg_cb)
 366   {
 367     tc_h->incoming_msg_cb (tc_h->cb_cls,
 368                            tc_h,
 369                            msg);
 370   }
 371   else
 372   {
 373     LOG (GNUNET_ERROR_TYPE_WARNING,
 374          "Incoming message from communicator but no handler!\n");
 375   }
 376   GNUNET_SERVICE_client_continue (tc_h->client);
 377 }
 378
 379
 380 /**
 381  * @brief Communicator informs that it tries to establish requested queue
 382  *
 383  * @param cls Closure - communicator handle
 384  * @param msg Message
 385  */
 386 static void
 387 handle_queue_create_ok (void *cls,
 388                         const struct GNUNET_TRANSPORT_CreateQueueResponse *msg)
 389 {
 390   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 391
 392   if (NULL != tc_h->queue_create_reply_cb)
 393   {
 394     tc_h->queue_create_reply_cb (tc_h->cb_cls, tc_h, GNUNET_YES);
 395   }
 396   GNUNET_SERVICE_client_continue (tc_h->client);
 397 }
 398
 399
 400 /**
 401  * @brief Communicator informs that it wont try establishing requested queue.
 402  *
 403  * It will not do so probably because the address is bougus (see comment to
 404  * #GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL)
 405  *
 406  * @param cls Closure - communicator handle
 407  * @param msg Message
 408  */
 409 static void
 410 handle_queue_create_fail (
 411   void *cls,
 412   const struct GNUNET_TRANSPORT_CreateQueueResponse *msg)
 413 {
 414   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 415
 416   if (NULL != tc_h->queue_create_reply_cb)
 417   {
 418     tc_h->queue_create_reply_cb (tc_h->cb_cls, tc_h, GNUNET_NO);
 419   }
 420   GNUNET_SERVICE_client_continue (tc_h->client);
 421 }
 422
 423
 424 /**
 425  * New queue became available.  Check message.
 426  *
 427  * @param cls the client
 428  * @param aqm the send message that was sent
 429  */
 430 static int
 431 check_add_queue_message (void *cls,
 432                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 433 {
 434   GNUNET_MQ_check_zero_termination (aqm);
 435   return GNUNET_OK;
 436 }
 437
 438
 439 /**
 440  * @brief Handle new queue
 441  *
 442  * Store context and call client callback.
 443  *
 444  * @param cls Closure - communicator handle
 445  * @param msg Message struct
 446  */
 447 static void
 448 handle_add_queue_message (void *cls,
 449                           const struct GNUNET_TRANSPORT_AddQueueMessage *msg)
 450 {
 451   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 452   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue;
 453
 454   tc_queue = tc_h->queue_head;
 455   if (NULL != tc_queue)
 456   {
 457     while (tc_queue->qid != msg->qid)
 458     {
 459       tc_queue = tc_queue->next;
 460     }
 461   } else {
 462     tc_queue =
 463       GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue);
 464     tc_queue->tc_h = tc_h;
 465     tc_queue->qid = msg->qid;
 466     tc_queue->peer_id = msg->receiver;
 467     GNUNET_CONTAINER_DLL_insert (tc_h->queue_head, tc_h->queue_tail, tc_queue);
 468   }
 469   GNUNET_assert (tc_queue->qid == msg->qid);
 470   GNUNET_assert (0 == GNUNET_memcmp (&tc_queue->peer_id, &msg->receiver));
 471   tc_queue->nt = msg->nt;
 472   tc_queue->mtu = msg->mtu;
 473   tc_queue->cs = msg->cs;
 474   if (NULL != tc_h->add_queue_cb)
 475   {
 476     tc_h->add_queue_cb (tc_h->cb_cls, tc_h, tc_queue);
 477   }
 478   GNUNET_SERVICE_client_continue (tc_h->client);
 479 }
 480
 481
 482 /**
 483  * @brief Shut down the service
 484  *
 485  * @param cls Closure - Handle to the service
 486  */
 487 static void
 488 shutdown_service (void *cls)
 489 {
 490   struct GNUNET_SERVICE_Handle *h = cls;
 491
 492   GNUNET_SERVICE_stop (h);
 493 }
 494
 495
 496 /**
 497  * @brief Callback called when new Client (Communicator) connects
 498  *
 499  * @param cls Closure - TransporCommmunicator Handle
 500  * @param client Client
 501  * @param mq Messagequeue
 502  *
 503  * @return TransportCommunicator Handle
 504  */
 505 static void *
 506 connect_cb (void *cls,
 507             struct GNUNET_SERVICE_Client *client,
 508             struct GNUNET_MQ_Handle *mq)
 509 {
 510   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 511   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue_iter;
 512
 513   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client connected.\n");
 514   tc_h->client = client;
 515   tc_h->c_mq = mq;
 516
 517   if (NULL == tc_h->queue_head)
 518     return tc_h;
 519   /* Iterate over queues. They are yet to be opened. Request opening. */
 520   while (NULL != (tc_queue_iter = tc_h->queue_head))
 521   {
 522     if (NULL == tc_queue_iter->open_queue_env)
 523       continue;
 524     /* Send the previously created mq envelope to request the creation of the
 525      * queue. */
 526     GNUNET_MQ_send (tc_h->c_mq, tc_queue_iter->open_queue_env);
 527     tc_queue_iter->open_queue_env = NULL;
 528   }
 529   return tc_h;
 530 }
 531
 532
 533 /**
 534  * @brief Callback called when Client disconnects
 535  *
 536  * @param cls Closure - TransportCommunicator Handle
 537  * @param client Client
 538  * @param internal_cls TransporCommmunicator Handle
 539  */
 540 static void
 541 disconnect_cb (void *cls,
 542                struct GNUNET_SERVICE_Client *client,
 543                void *internal_cls)
 544 {
 545   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 546
 547   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client disconnected.\n");
 548   tc_h->client = NULL;
 549 }
 550
 551
 552 /**
 553  * @brief Start the communicator part of the transport service
 554  *
 555  * @param communicator_available Callback to be called when a new communicator
 556  * becomes available
 557  * @param cfg Configuration
 558  */
 559 static void
 560 transport_communicator_start (
 561   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h)
 562 {
 563   struct GNUNET_MQ_MessageHandler mh[] = {
 564     GNUNET_MQ_hd_var_size (communicator_available,
 565                            GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
 566                            struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
 567                            &tc_h),
 568     // GNUNET_MQ_hd_var_size (communicator_backchannel,
 569     //    GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL,
 570     //    struct GNUNET_TRANSPORT_CommunicatorBackchannel,
 571     //    NULL),
 572     GNUNET_MQ_hd_var_size (add_address,
 573                            GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
 574                            struct GNUNET_TRANSPORT_AddAddressMessage,
 575                            &tc_h),
 576     // GNUNET_MQ_hd_fixed_size (del_address,
 577     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
 578     //                         struct GNUNET_TRANSPORT_DelAddressMessage,
 579     //                         NULL),
 580     GNUNET_MQ_hd_var_size (incoming_msg,
 581                            GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
 582                            struct GNUNET_TRANSPORT_IncomingMessage,
 583                            NULL),
 584     GNUNET_MQ_hd_fixed_size (queue_create_ok,
 585                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK,
 586                              struct GNUNET_TRANSPORT_CreateQueueResponse,
 587                              tc_h),
 588     GNUNET_MQ_hd_fixed_size (queue_create_fail,
 589                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL,
 590                              struct GNUNET_TRANSPORT_CreateQueueResponse,
 591                              tc_h),
 592     GNUNET_MQ_hd_var_size (add_queue_message,
 593                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
 594                            struct GNUNET_TRANSPORT_AddQueueMessage,
 595                            NULL),
 596     // GNUNET_MQ_hd_fixed_size (del_queue_message,
 597     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
 598     //                         struct GNUNET_TRANSPORT_DelQueueMessage,
 599     //                         NULL),
 600     // GNUNET_MQ_hd_fixed_size (send_message_ack,
 601     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
 602     //                         struct GNUNET_TRANSPORT_SendMessageToAck,
 603     //                         NULL),
 604   };
 605   struct GNUNET_SERVICE_Handle *h;
 606
 607   h = GNUNET_SERVICE_start ("transport",
 608                             tc_h->cfg,
 609                             &connect_cb,
 610                             &disconnect_cb,
 611                             tc_h,
 612                             mh);
 613   if (NULL == h)
 614     LOG (GNUNET_ERROR_TYPE_ERROR, "Failed starting service!\n");
 615   else
 616   {
 617     LOG (GNUNET_ERROR_TYPE_DEBUG, "Started service\n");
 618     /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_service, h);
 619   }
 620 }
 621
 622
 623 /**
 624  * @brief Task run at shutdown to kill communicator and clean up
 625  *
 626  * @param cls Closure - Process of communicator
 627  */
 628 static void
 629 shutdown_communicator (void *cls)
 630 {
 631   struct GNUNET_OS_Process *proc = cls;
 632
 633   if (GNUNET_OK != GNUNET_OS_process_kill (proc, SIGTERM))
 634   {
 635     LOG (GNUNET_ERROR_TYPE_WARNING,
 636          "Error shutting down communicator with SIGERM, trying SIGKILL\n");
 637     if (GNUNET_OK != GNUNET_OS_process_kill (proc, SIGKILL))
 638     {
 639       LOG (GNUNET_ERROR_TYPE_ERROR,
 640            "Error shutting down communicator with SIGERM and SIGKILL\n");
 641     }
 642   }
 643   GNUNET_OS_process_destroy (proc);
 644 }
 645
 646
 647 /**
 648  * @brief Start the communicator
 649  *
 650  * @param cfgname Name of the communicator
 651  */
 652 static void
 653 communicator_start (
 654   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 655   const char *binary_name)
 656 {
 657   char *binary;
 658
 659   LOG (GNUNET_ERROR_TYPE_DEBUG, "communicator_start\n");
 660   binary = GNUNET_OS_get_libexec_binary_path (binary_name);
 661   tc_h->c_proc = GNUNET_OS_start_process (GNUNET_YES,
 662                                           GNUNET_OS_INHERIT_STD_OUT_AND_ERR,
 663                                           NULL,
 664                                           NULL,
 665                                           NULL,
 666                                           binary,
 667                                           binary_name,
 668                                           "-c",
 669                                           tc_h->cfg_filename,
 670                                           NULL);
 671   if (NULL == tc_h->c_proc)
 672   {
 673     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to start communicator!");
 674     return;
 675   }
 676   LOG (GNUNET_ERROR_TYPE_INFO, "started communicator\n");
 677   GNUNET_free (binary);
 678   /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_communicator,
 679                                             tc_h->c_proc);
 680 }
 681
 682
 683 /**
 684  * @brief Start communicator part of transport service and communicator
 685  *
 686  * @param service_name Name of the service
 687  * @param cfg Configuration handle
 688  * @param communicator_available_cb Callback that is called when a new
 689  * @param add_address_cb Callback that is called when a new
 690  * communicator becomes available
 691  * @param cb_cls Closure to @a communicator_available_cb and @a
 692  *
 693  * @return Handle to the communicator duo
 694  */
 695 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *
 696 GNUNET_TRANSPORT_TESTING_transport_communicator_service_start (
 697   const char *service_name,
 698   const char *binary_name,
 699   const char *cfg_filename,
 700   GNUNET_TRANSPORT_TESTING_CommunicatorAvailableCallback
 701   communicator_available_cb,
 702   GNUNET_TRANSPORT_TESTING_AddAddressCallback add_address_cb,
 703   GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback queue_create_reply_cb,
 704   GNUNET_TRANSPORT_TESTING_AddQueueCallback add_queue_cb,
 705   GNUNET_TRANSPORT_TESTING_IncomingMessageCallback incoming_message_cb,
 706   void *cb_cls)
 707 {
 708   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h;
 709
 710   tc_h =
 711     GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle);
 712   tc_h->cfg_filename = GNUNET_strdup (cfg_filename);
 713   tc_h->cfg = GNUNET_CONFIGURATION_create ();
 714   if ((GNUNET_SYSERR == GNUNET_CONFIGURATION_load (tc_h->cfg, cfg_filename)))
 715   {
 716     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 717                 _ ("Malformed configuration file `%s', exit ...\n"),
 718                 cfg_filename);
 719     GNUNET_free (tc_h->cfg_filename);
 720     GNUNET_CONFIGURATION_destroy (tc_h->cfg);
 721     GNUNET_free (tc_h);
 722     return NULL;
 723   }
 724   tc_h->communicator_available_cb = communicator_available_cb;
 725   tc_h->add_address_cb = add_address_cb;
 726   tc_h->queue_create_reply_cb = queue_create_reply_cb;
 727   tc_h->add_queue_cb = add_queue_cb;
 728   tc_h->incoming_msg_cb = incoming_message_cb;
 729   tc_h->cb_cls = cb_cls;
 730
 731   /* Start communicator part of service */
 732   transport_communicator_start (tc_h);
 733
 734   /* Schedule start communicator */
 735   communicator_start (tc_h,
 736                       binary_name);
 737   return tc_h;
 738 }
 739
 740
 741 /**
 742  * @brief Instruct communicator to open a queue
 743  *
 744  * @param tc_h Handle to communicator which shall open queue
 745  * @param peer_id Towards which peer
 746  * @param address For which address
 747  */
 748 void
 749 GNUNET_TRANSPORT_TESTING_transport_communicator_open_queue (
 750   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 751   const struct GNUNET_PeerIdentity *peer_id,
 752   const char *address)
 753 {
 754   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue;
 755   static uint32_t idgen;
 756   char *prefix;
 757   struct GNUNET_TRANSPORT_CreateQueue *msg;
 758   struct GNUNET_MQ_Envelope *env;
 759   size_t alen;
 760
 761   tc_queue =
 762     GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue);
 763   tc_queue->tc_h = tc_h;
 764   prefix = GNUNET_HELLO_address_to_prefix (address);
 765   if (NULL == prefix)
 766   {
 767     GNUNET_break (0);  /* We got an invalid address!? */
 768     GNUNET_free (tc_queue);
 769     return;
 770   }
 771   GNUNET_free (prefix);
 772   alen = strlen (address) + 1;
 773   env =
 774     GNUNET_MQ_msg_extra (msg, alen, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE);
 775   msg->request_id = htonl (idgen++);
 776   tc_queue->qid = msg->request_id;
 777   msg->receiver = *peer_id;
 778   tc_queue->peer_id = *peer_id;
 779   memcpy (&msg[1], address, alen);
 780   if (NULL != tc_h->c_mq)
 781   {
 782     GNUNET_MQ_send (tc_h->c_mq, env);
 783   }
 784   else
 785   {
 786     tc_queue->open_queue_env = env;
 787   }
 788   GNUNET_CONTAINER_DLL_insert (tc_h->queue_head, tc_h->queue_tail, tc_queue);
 789 }
 790
 791
 792 /**
 793  * @brief Instruct communicator to send data
 794  *
 795  * @param tc_queue The queue to use for sending
 796  * @param payload Data to send
 797  * @param payload_size Size of the payload
 798  *
 799  * @return Handle to the transmission
 800  */
 801 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *
 802 GNUNET_TRANSPORT_TESTING_transport_communicator_send
 803   (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue,
 804   const void *payload,
 805   size_t payload_size /*,
 806                          GNUNET_TRANSPORT_TESTING_SuccessStatus cb,
 807                          void *cb_cls*/)
 808 {
 809   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *tc_t;
 810   struct GNUNET_MessageHeader *mh;
 811   struct GNUNET_TRANSPORT_SendMessageTo *msg;
 812   struct GNUNET_MQ_Envelope *env;
 813   size_t inbox_size;
 814
 815   inbox_size = sizeof(struct GNUNET_MessageHeader) + payload_size;
 816   mh = GNUNET_malloc (inbox_size);
 817   mh->size = htons (inbox_size);
 818   mh->type = GNUNET_MESSAGE_TYPE_DUMMY;
 819   memcpy (&mh[1],
 820           payload,
 821           payload_size);
 822   env = GNUNET_MQ_msg_extra (msg,
 823                              inbox_size,
 824                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
 825   msg->qid = htonl (tc_queue->qid);
 826   msg->mid = tc_queue->mid++;
 827   msg->receiver = tc_queue->peer_id;
 828   memcpy (&msg[1], mh, inbox_size);
 829   GNUNET_free (mh);
 830   GNUNET_MQ_send (tc_queue->tc_h->c_mq, env);
 831   return tc_t;
 832 }