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
 310   size = ntohs (msg->header.size) - sizeof(*msg);
 311   if (0 == size)
 312     return; /* receive-only communicator */
 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 communicator_available()\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                            (const struct GNUNET_MessageHeader *) 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   while (tc_queue->qid != msg->qid)
 456   {
 457     tc_queue = tc_queue->next;
 458   }
 459   GNUNET_assert (tc_queue->qid == msg->qid);
 460   GNUNET_assert (0 == GNUNET_memcmp (&tc_queue->peer_id, &msg->receiver));
 461   tc_queue->nt = msg->nt;
 462   tc_queue->mtu = msg->mtu;
 463   tc_queue->cs = msg->cs;
 464   if (NULL != tc_h->add_queue_cb)
 465   {
 466     tc_h->add_queue_cb (tc_h->cb_cls, tc_h, tc_queue);
 467   }
 468   GNUNET_SERVICE_client_continue (tc_h->client);
 469 }
 470
 471
 472 /**
 473  * @brief Shut down the service
 474  *
 475  * @param cls Closure - Handle to the service
 476  */
 477 static void
 478 shutdown_service (void *cls)
 479 {
 480   struct GNUNET_SERVICE_Handle *h = cls;
 481
 482   GNUNET_SERVICE_stop (h);
 483 }
 484
 485
 486 /**
 487  * @brief Callback called when new Client (Communicator) connects
 488  *
 489  * @param cls Closure - TransporCommmunicator Handle
 490  * @param client Client
 491  * @param mq Messagequeue
 492  *
 493  * @return TransportCommunicator Handle
 494  */
 495 static void *
 496 connect_cb (void *cls,
 497             struct GNUNET_SERVICE_Client *client,
 498             struct GNUNET_MQ_Handle *mq)
 499 {
 500   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 501   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue_iter;
 502
 503   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client connected.\n");
 504   tc_h->client = client;
 505   tc_h->c_mq = mq;
 506
 507   if (NULL == tc_h->queue_head)
 508     return tc_h;
 509   /* Iterate over queues. They are yet to be opened. Request opening. */
 510   while (NULL != (tc_queue_iter = tc_h->queue_head))
 511   {
 512     if (NULL == tc_queue_iter->open_queue_env)
 513       continue;
 514     /* Send the previously created mq envelope to request the creation of the
 515      * queue. */
 516     GNUNET_MQ_send (tc_h->c_mq, tc_queue_iter->open_queue_env);
 517     tc_queue_iter->open_queue_env = NULL;
 518   }
 519   return tc_h;
 520 }
 521
 522
 523 /**
 524  * @brief Callback called when Client disconnects
 525  *
 526  * @param cls Closure - TransportCommunicator Handle
 527  * @param client Client
 528  * @param internal_cls TransporCommmunicator Handle
 529  */
 530 static void
 531 disconnect_cb (void *cls,
 532                struct GNUNET_SERVICE_Client *client,
 533                void *internal_cls)
 534 {
 535   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 536
 537   LOG (GNUNET_ERROR_TYPE_DEBUG, "Client disconnected.\n");
 538   tc_h->client = NULL;
 539 }
 540
 541
 542 /**
 543  * @brief Start the communicator part of the transport service
 544  *
 545  * @param communicator_available Callback to be called when a new communicator
 546  * becomes available
 547  * @param cfg Configuration
 548  */
 549 static void
 550 transport_communicator_start (
 551   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h)
 552 {
 553   struct GNUNET_MQ_MessageHandler mh[] = {
 554     GNUNET_MQ_hd_var_size (communicator_available,
 555                            GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
 556                            struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
 557                            &tc_h),
 558     // GNUNET_MQ_hd_var_size (communicator_backchannel,
 559     //    GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL,
 560     //    struct GNUNET_TRANSPORT_CommunicatorBackchannel,
 561     //    NULL),
 562     GNUNET_MQ_hd_var_size (add_address,
 563                            GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
 564                            struct GNUNET_TRANSPORT_AddAddressMessage,
 565                            &tc_h),
 566     // GNUNET_MQ_hd_fixed_size (del_address,
 567     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
 568     //                         struct GNUNET_TRANSPORT_DelAddressMessage,
 569     //                         NULL),
 570     GNUNET_MQ_hd_var_size (incoming_msg,
 571                            GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
 572                            struct GNUNET_TRANSPORT_IncomingMessage,
 573                            NULL),
 574     GNUNET_MQ_hd_fixed_size (queue_create_ok,
 575                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK,
 576                              struct GNUNET_TRANSPORT_CreateQueueResponse,
 577                              tc_h),
 578     GNUNET_MQ_hd_fixed_size (queue_create_fail,
 579                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL,
 580                              struct GNUNET_TRANSPORT_CreateQueueResponse,
 581                              tc_h),
 582     GNUNET_MQ_hd_var_size (add_queue_message,
 583                            GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
 584                            struct GNUNET_TRANSPORT_AddQueueMessage,
 585                            NULL),
 586     // GNUNET_MQ_hd_fixed_size (del_queue_message,
 587     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
 588     //                         struct GNUNET_TRANSPORT_DelQueueMessage,
 589     //                         NULL),
 590     // GNUNET_MQ_hd_fixed_size (send_message_ack,
 591     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
 592     //                         struct GNUNET_TRANSPORT_SendMessageToAck,
 593     //                         NULL),
 594   };
 595   struct GNUNET_SERVICE_Handle *h;
 596
 597   h = GNUNET_SERVICE_start ("transport",
 598                             tc_h->cfg,
 599                             &connect_cb,
 600                             &disconnect_cb,
 601                             tc_h,
 602                             mh);
 603   if (NULL == h)
 604     LOG (GNUNET_ERROR_TYPE_ERROR, "Failed starting service!\n");
 605   else
 606   {
 607     LOG (GNUNET_ERROR_TYPE_DEBUG, "Started service\n");
 608     /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_service, h);
 609   }
 610 }
 611
 612
 613 /**
 614  * @brief Task run at shutdown to kill communicator and clean up
 615  *
 616  * @param cls Closure - Process of communicator
 617  */
 618 static void
 619 shutdown_communicator (void *cls)
 620 {
 621   struct GNUNET_OS_Process *proc = cls;
 622
 623   if (GNUNET_OK != GNUNET_OS_process_kill (proc, SIGTERM))
 624   {
 625     LOG (GNUNET_ERROR_TYPE_WARNING,
 626          "Error shutting down communicator with SIGERM, trying SIGKILL\n");
 627     if (GNUNET_OK != GNUNET_OS_process_kill (proc, SIGKILL))
 628     {
 629       LOG (GNUNET_ERROR_TYPE_ERROR,
 630            "Error shutting down communicator with SIGERM and SIGKILL\n");
 631     }
 632   }
 633   GNUNET_OS_process_destroy (proc);
 634 }
 635
 636
 637 /**
 638  * @brief Start the communicator
 639  *
 640  * @param cfgname Name of the communicator
 641  */
 642 static void
 643 communicator_start (
 644   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 645   const char *binary_name)
 646 {
 647   char *binary;
 648
 649   LOG (GNUNET_ERROR_TYPE_DEBUG, "communicator_start\n");
 650   binary = GNUNET_OS_get_libexec_binary_path (binary_name);
 651   tc_h->c_proc = GNUNET_OS_start_process (GNUNET_YES,
 652                                           GNUNET_OS_INHERIT_STD_OUT_AND_ERR,
 653                                           NULL,
 654                                           NULL,
 655                                           NULL,
 656                                           binary,
 657                                           binary_name,
 658                                           "-c",
 659                                           tc_h->cfg_filename,
 660                                           NULL);
 661   if (NULL == tc_h->c_proc)
 662   {
 663     GNUNET_log (GNUNET_ERROR_TYPE_ERROR, "Failed to start communicator!");
 664     return;
 665   }
 666   LOG (GNUNET_ERROR_TYPE_DEBUG, "started communicator\n");
 667   GNUNET_free (binary);
 668   /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_communicator,
 669                                             tc_h->c_proc);
 670 }
 671
 672
 673 /**
 674  * @brief Start communicator part of transport service and communicator
 675  *
 676  * @param service_name Name of the service
 677  * @param cfg Configuration handle
 678  * @param communicator_available_cb Callback that is called when a new
 679  * @param add_address_cb Callback that is called when a new
 680  * communicator becomes available
 681  * @param cb_cls Closure to @a communicator_available_cb and @a
 682  *
 683  * @return Handle to the communicator duo
 684  */
 685 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *
 686 GNUNET_TRANSPORT_TESTING_transport_communicator_service_start (
 687   const char *service_name,
 688   const char *binary_name,
 689   const char *cfg_filename,
 690   GNUNET_TRANSPORT_TESTING_CommunicatorAvailableCallback
 691   communicator_available_cb,
 692   GNUNET_TRANSPORT_TESTING_AddAddressCallback add_address_cb,
 693   GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback queue_create_reply_cb,
 694   GNUNET_TRANSPORT_TESTING_AddQueueCallback add_queue_cb,
 695   GNUNET_TRANSPORT_TESTING_IncomingMessageCallback incoming_message_cb,
 696   void *cb_cls)
 697 {
 698   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h;
 699
 700   tc_h =
 701     GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle);
 702   tc_h->cfg_filename = GNUNET_strdup (cfg_filename);
 703   tc_h->cfg = GNUNET_CONFIGURATION_create ();
 704   if ((GNUNET_SYSERR == GNUNET_CONFIGURATION_load (tc_h->cfg, cfg_filename)))
 705   {
 706     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 707                 _ ("Malformed configuration file `%s', exit ...\n"),
 708                 cfg_filename);
 709     GNUNET_free (tc_h->cfg_filename);
 710     GNUNET_CONFIGURATION_destroy (tc_h->cfg);
 711     GNUNET_free (tc_h);
 712     return NULL;
 713   }
 714   tc_h->communicator_available_cb = communicator_available_cb;
 715   tc_h->add_address_cb = add_address_cb;
 716   tc_h->queue_create_reply_cb = queue_create_reply_cb;
 717   tc_h->add_queue_cb = add_queue_cb;
 718   tc_h->incoming_msg_cb = incoming_message_cb;
 719   tc_h->cb_cls = cb_cls;
 720
 721   /* Start communicator part of service */
 722   transport_communicator_start (tc_h);
 723
 724   /* Schedule start communicator */
 725   communicator_start (tc_h,
 726                       binary_name);
 727   return tc_h;
 728 }
 729
 730
 731 /**
 732  * @brief Instruct communicator to open a queue
 733  *
 734  * @param tc_h Handle to communicator which shall open queue
 735  * @param peer_id Towards which peer
 736  * @param address For which address
 737  */
 738 void
 739 GNUNET_TRANSPORT_TESTING_transport_communicator_open_queue (
 740   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 741   const struct GNUNET_PeerIdentity *peer_id,
 742   const char *address)
 743 {
 744   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue;
 745   static uint32_t idgen;
 746   char *prefix;
 747   struct GNUNET_TRANSPORT_CreateQueue *msg;
 748   struct GNUNET_MQ_Envelope *env;
 749   size_t alen;
 750
 751   tc_queue =
 752     GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue);
 753   tc_queue->tc_h = tc_h;
 754   prefix = GNUNET_HELLO_address_to_prefix (address);
 755   if (NULL == prefix)
 756   {
 757     GNUNET_break (0);  /* We got an invalid address!? */
 758     GNUNET_free (tc_queue);
 759     return;
 760   }
 761   GNUNET_free (prefix);
 762   alen = strlen (address) + 1;
 763   env =
 764     GNUNET_MQ_msg_extra (msg, alen, GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE);
 765   msg->request_id = htonl (idgen++);
 766   tc_queue->qid = msg->request_id;
 767   msg->receiver = *peer_id;
 768   tc_queue->peer_id = *peer_id;
 769   memcpy (&msg[1], address, alen);
 770   if (NULL != tc_h->c_mq)
 771   {
 772     GNUNET_MQ_send (tc_h->c_mq, env);
 773   }
 774   else
 775   {
 776     tc_queue->open_queue_env = env;
 777   }
 778   GNUNET_CONTAINER_DLL_insert (tc_h->queue_head, tc_h->queue_tail, tc_queue);
 779 }
 780
 781
 782 /**
 783  * @brief Instruct communicator to send data
 784  *
 785  * @param tc_queue The queue to use for sending
 786  * @param payload Data to send
 787  * @param payload_size Size of the payload
 788  *
 789  * @return Handle to the transmission
 790  */
 791 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *
 792 GNUNET_TRANSPORT_TESTING_transport_communicator_send
 793   (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue,
 794   const void *payload,
 795   size_t payload_size /*,
 796                          GNUNET_TRANSPORT_TESTING_SuccessStatus cb,
 797                          void *cb_cls*/)
 798 {
 799   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *tc_t;
 800   struct GNUNET_MessageHeader *mh;
 801   struct GNUNET_TRANSPORT_SendMessageTo *msg;
 802   struct GNUNET_MQ_Envelope *env;
 803   size_t inbox_size;
 804
 805   inbox_size = sizeof(struct GNUNET_MessageHeader) + payload_size;
 806   mh = GNUNET_malloc (inbox_size);
 807   mh->size = htons (inbox_size);
 808   mh->type = GNUNET_MESSAGE_TYPE_DUMMY;
 809   memcpy (&mh[1],
 810           payload,
 811           payload_size);
 812   env = GNUNET_MQ_msg_extra (msg,
 813                              inbox_size,
 814                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
 815   msg->qid = htonl (tc_queue->qid);
 816   msg->mid = tc_queue->mid++;
 817   msg->receiver = tc_queue->peer_id;
 818   memcpy (&msg[1], mh, inbox_size);
 819   GNUNET_free (mh);
 820   GNUNET_MQ_send (tc_queue->tc_h->c_mq, env);
 821   return tc_t;
 822 }