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 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle
  42 {
  43   /**
  44    * @brief Handle to the configuration
  45    */
  46   struct GNUNET_CONFIGURATION_Handle *cfg;
  47
  48   /**
  49    * @brief File name of configuration file
  50    */
  51   char *cfg_filename;
  52
  53   /**
  54    * @brief Handle to the transport service
  55    */
  56   struct GNUNET_SERVICE_Handle *tsh;
  57
  58   /**
  59    * @brief Task that will be run on shutdown to stop and clean transport
  60    * service
  61    */
  62   struct GNUNET_SCHEDULER_Task *ts_shutdown_task;
  63
  64   /**
  65    * @brief Handle to the client
  66    */
  67   struct GNUNET_SERVICE_Client *client;
  68
  69   /**
  70    * @brief Handle to the client
  71    */
  72   struct GNUNET_MQ_Handle *c_mq;
  73
  74   /**
  75    * @brief Process of the communicator
  76    */
  77   struct GNUNET_OS_Process *c_proc;
  78
  79   /**
  80    * @brief Task that will be run on shutdown to stop and clean communicator
  81    */
  82   struct GNUNET_SCHEDULER_Task *c_shutdown_task;
  83
  84   /**
  85    * @brief Characteristics of the communicator
  86    */
  87   enum GNUNET_TRANSPORT_CommunicatorCharacteristics c_characteristics;
  88
  89   /**
  90    * @brief Specifies supported addresses
  91    */
  92   char *c_addr_prefix;
  93
  94   /**
  95    * @brief Specifies supported addresses
  96    */
  97   char *c_address;
  98
  99   /**
 100    * @brief Head of the queues
 101    */
 102   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *queue_head;
 103
 104   /**
 105    * @brief Tail of the queues
 106    */
 107   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *queue_tail;
 108
 109   /* Callbacks + Closures */
 110   /**
 111    * @brief Callback called when a new communicator connects
 112    */
 113   GNUNET_TRANSPORT_TESTING_CommunicatorAvailableCallback communicator_available_cb;
 114
 115   /**
 116    * @brief Callback called when a new communicator connects
 117    */
 118   GNUNET_TRANSPORT_TESTING_AddAddressCallback add_address_cb;
 119
 120   /**
 121    * @brief Callback called when a new communicator connects
 122    */
 123   GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback queue_create_reply_cb;
 124
 125   /**
 126    * @brief Callback called when a new communicator connects
 127    */
 128   GNUNET_TRANSPORT_TESTING_AddQueueCallback add_queue_cb;
 129
 130   /**
 131    * @brief Closure to the callback
 132    */
 133   void *cb_cls;
 134 };
 135
 136
 137 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue
 138 {
 139   /**
 140    * @brief Handle to the TransportCommunicator
 141    */
 142   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h;
 143
 144   /**
 145    * @brief Task to request the opening of a view
 146    */
 147   struct GNUNET_MQ_Envelope *open_queue_env;
 148
 149   /**
 150    * @brief Peer ID of the peer on the other side of the queue
 151    */
 152   struct GNUNET_PeerIdentity peer_id;
 153
 154   /**
 155    * @brief Queue ID
 156    */
 157   uint32_t qid;
 158
 159   /**
 160    * @brief Current message id
 161    */
 162   uint64_t mid;
 163
 164   /**
 165    * An `enum GNUNET_NetworkType` in NBO.
 166    */
 167   uint32_t nt;
 168
 169   /**
 170    * Maximum transmission unit, in NBO.  UINT32_MAX for unlimited.
 171    */
 172   uint32_t mtu;
 173
 174   /**
 175    * An `enum GNUNET_TRANSPORT_ConnectionStatus` in NBO.
 176    */
 177   uint32_t cs;
 178
 179   /**
 180    * @brief Next element inside a DLL
 181    */
 182   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *next;
 183
 184   /**
 185    * @brief Previous element inside a DLL
 186    */
 187   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *prev;
 188 };
 189
 190
 191 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission
 192 {
 193 };
 194
 195
 196 /**
 197  * @brief Check whether incoming msg indicating available communicator is
 198  * correct
 199  *
 200  * @param cls Closure
 201  * @param msg Message struct
 202  *
 203  * @return GNUNET_YES in case message is correct
 204  */
 205 static int
 206 check_communicator_available (void *cls,
 207     const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *msg)
 208 {
 209   uint16_t size;
 210
 211   size = ntohs (msg->header.size) - sizeof (*msg);
 212   if (0 == size)
 213     return GNUNET_OK; /* receive-only communicator */
 214   GNUNET_MQ_check_zero_termination (msg);
 215   return GNUNET_OK;
 216 }
 217
 218
 219 /**
 220  * @brief Handle new communicator
 221  *
 222  * @param cls Closure
 223  * @param msg Message struct
 224  */
 225 static void
 226 handle_communicator_available (void *cls,
 227     const struct GNUNET_TRANSPORT_CommunicatorAvailableMessage *msg)
 228 {
 229   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 230   uint16_t size;
 231
 232   size = ntohs (msg->header.size) - sizeof (*msg);
 233   if (0 == size)
 234     return; /* receive-only communicator */
 235   tc_h->c_characteristics = ntohl (msg->cc);
 236   tc_h->c_addr_prefix = GNUNET_strdup ((const char *) &msg[1]);
 237   if (NULL != tc_h->communicator_available_cb)
 238   {
 239     LOG (GNUNET_ERROR_TYPE_DEBUG,
 240         "calling communicator_available_cb()\n");
 241     tc_h->communicator_available_cb (tc_h->cb_cls,
 242                                      tc_h,
 243                                      tc_h->c_characteristics,
 244                                      tc_h->c_addr_prefix);
 245   }
 246   GNUNET_SERVICE_client_continue (tc_h->client);
 247 }
 248
 249
 250 /**
 251  * Address of our peer added.  Test message is well-formed.
 252  *
 253  * @param cls the client
 254  * @param aam the send message that was sent
 255  * @return #GNUNET_OK if message is well-formed
 256  */
 257 static int
 258 check_add_address (void *cls,
 259                    const struct GNUNET_TRANSPORT_AddAddressMessage *msg)
 260 {
 261   struct TransportClient *tc = cls;
 262
 263   //if (CT_COMMUNICATOR != tc->type)
 264   //{
 265   //  GNUNET_break (0);
 266   //  return GNUNET_SYSERR;
 267   //}
 268   GNUNET_MQ_check_zero_termination (msg);
 269   return GNUNET_OK;
 270 }
 271
 272
 273 static void
 274 handle_add_address (void *cls,
 275                     const struct GNUNET_TRANSPORT_AddAddressMessage *msg)
 276 {
 277   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 278   uint16_t size;
 279
 280   size = ntohs (msg->header.size) - sizeof (*msg);
 281   if (0 == size)
 282     return; /* receive-only communicator */
 283   tc_h->c_address = GNUNET_strdup ((const char *) &msg[1]);
 284   if (NULL != tc_h->add_address_cb)
 285   {
 286     LOG (GNUNET_ERROR_TYPE_DEBUG,
 287         "calling communicator_available()\n");
 288     tc_h->add_address_cb (tc_h->cb_cls,
 289                           tc_h,
 290                           tc_h->c_address,
 291                           GNUNET_TIME_relative_ntoh (msg->expiration),
 292                           msg->aid,
 293                           ntohl (msg->nt));
 294   }
 295   GNUNET_SERVICE_client_continue (tc_h->client);
 296 }
 297
 298
 299 static void
 300 handle_queue_create_ok (void *cls,
 301                         const struct GNUNET_TRANSPORT_CreateQueueResponse *msg)
 302 {
 303   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 304
 305   if (NULL != tc_h->queue_create_reply_cb)
 306   {
 307     tc_h->queue_create_reply_cb (tc_h->cb_cls,
 308                             tc_h,
 309                             GNUNET_YES);
 310   }
 311   GNUNET_SERVICE_client_continue (tc_h->client);
 312 }
 313
 314
 315 static void
 316 handle_queue_create_fail (void *cls,
 317                           const struct GNUNET_TRANSPORT_CreateQueueResponse *msg)
 318 {
 319   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 320
 321   if (NULL != tc_h->queue_create_reply_cb)
 322   {
 323     tc_h->queue_create_reply_cb (tc_h->cb_cls,
 324                             tc_h,
 325                             GNUNET_NO);
 326   }
 327   GNUNET_SERVICE_client_continue (tc_h->client);
 328 }
 329
 330
 331 /**
 332  * New queue became available.  Check message.
 333  *
 334  * @param cls the client
 335  * @param aqm the send message that was sent
 336  */
 337 static int
 338 check_add_queue_message (void *cls,
 339                          const struct GNUNET_TRANSPORT_AddQueueMessage *aqm)
 340 {
 341   GNUNET_MQ_check_zero_termination (aqm);
 342   return GNUNET_OK;
 343 }
 344
 345
 346 /**
 347  * @brief Handle new communicator
 348  *
 349  * @param cls Closure
 350  * @param msg Message struct
 351  */
 352 static void
 353 handle_add_queue_message (void *cls,
 354                           const struct GNUNET_TRANSPORT_AddQueueMessage *msg)
 355 {
 356   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 357   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue;
 358
 359   tc_queue = tc_h->queue_head;
 360   while (tc_queue->qid != msg->qid)
 361   {
 362     tc_queue = tc_queue->next;
 363   }
 364   GNUNET_assert (tc_queue->qid == msg->qid);
 365   GNUNET_assert (0 == GNUNET_memcmp (&tc_queue->peer_id,
 366                                      &msg->receiver));
 367   tc_queue->nt = msg->nt;
 368   tc_queue->mtu = msg->mtu;
 369   tc_queue->cs = msg->cs;
 370   if (NULL != tc_h->add_queue_cb)
 371   {
 372     tc_h->add_queue_cb (tc_h->cb_cls,
 373                         tc_h,
 374                         tc_queue);
 375   }
 376   GNUNET_SERVICE_client_continue (tc_h->client);
 377 }
 378
 379
 380 /**
 381  * @brief Shut down the service
 382  *
 383  * @param cls Closure - Handle to the service
 384  */
 385 static void
 386 shutdown_service (void *cls)
 387 {
 388   struct GNUNET_SERVICE_Handle *h = cls;
 389
 390   GNUNET_SERVICE_stop (h);
 391 }
 392
 393
 394 /**
 395  * @brief Callback called when new Client (Communicator) connects
 396  *
 397  * @param cls Closure - TransporCommmunicator Handle
 398  * @param client Client
 399  * @param mq Messagequeue
 400  *
 401  * @return TransportCommunicator Handle
 402  */
 403 static void *
 404 connect_cb (void *cls,
 405             struct GNUNET_SERVICE_Client *client,
 406             struct GNUNET_MQ_Handle *mq)
 407 {
 408   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 409   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue_iter;
 410
 411   LOG (GNUNET_ERROR_TYPE_DEBUG,
 412       "Client connected.\n");
 413   tc_h->client = client;
 414   tc_h->c_mq = mq;
 415
 416   if (NULL == tc_h->queue_head) return tc_h;
 417   while (NULL != (tc_queue_iter = tc_h->queue_head))
 418   {
 419     if (NULL == tc_queue_iter->open_queue_env) continue;
 420     GNUNET_MQ_send (tc_h->c_mq,
 421                     tc_queue_iter->open_queue_env);
 422     tc_queue_iter->open_queue_env = NULL;
 423   }
 424   return tc_h;
 425 }
 426
 427
 428 /**
 429  * @brief Callback called when Client disconnects
 430  *
 431  * @param cls Closure - TransportCommunicator Handle
 432  * @param client Client
 433  * @param internal_cls TransporCommmunicator Handle
 434  */
 435 static void
 436 disconnect_cb (void *cls,
 437                struct GNUNET_SERVICE_Client *client,
 438                void *internal_cls)
 439 {
 440   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h = cls;
 441
 442   LOG (GNUNET_ERROR_TYPE_DEBUG,
 443       "Client disconnected.\n");
 444   tc_h->client = NULL;
 445 }
 446
 447
 448 /**
 449  * @brief Start the communicator part of the transport service
 450  *
 451  * @param communicator_available Callback to be called when a new communicator
 452  * becomes available
 453  * @param cfg Configuration
 454  */
 455 static void
 456 transport_communicator_start (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h)
 457 {
 458   struct GNUNET_MQ_MessageHandler mh[] = {
 459     GNUNET_MQ_hd_var_size (communicator_available,
 460         GNUNET_MESSAGE_TYPE_TRANSPORT_NEW_COMMUNICATOR,
 461         struct GNUNET_TRANSPORT_CommunicatorAvailableMessage,
 462         &tc_h),
 463     //GNUNET_MQ_hd_var_size (communicator_backchannel,
 464     //    GNUNET_MESSAGE_TYPE_TRANSPORT_COMMUNICATOR_BACKCHANNEL,
 465     //    struct GNUNET_TRANSPORT_CommunicatorBackchannel,
 466     //    NULL),
 467     GNUNET_MQ_hd_var_size (add_address,
 468         GNUNET_MESSAGE_TYPE_TRANSPORT_ADD_ADDRESS,
 469         struct GNUNET_TRANSPORT_AddAddressMessage,
 470         &tc_h),
 471     //GNUNET_MQ_hd_fixed_size (del_address,
 472     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_DEL_ADDRESS,
 473     //                         struct GNUNET_TRANSPORT_DelAddressMessage,
 474     //                         NULL),
 475     //GNUNET_MQ_hd_var_size (incoming_msg,
 476     //    GNUNET_MESSAGE_TYPE_TRANSPORT_INCOMING_MSG,
 477     //    struct GNUNET_TRANSPORT_IncomingMessage,
 478     //    NULL),
 479     GNUNET_MQ_hd_fixed_size (queue_create_ok,
 480           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_OK,
 481           struct GNUNET_TRANSPORT_CreateQueueResponse,
 482           tc_h),
 483     GNUNET_MQ_hd_fixed_size (queue_create_fail,
 484           GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE_FAIL,
 485           struct GNUNET_TRANSPORT_CreateQueueResponse,
 486           tc_h),
 487     GNUNET_MQ_hd_var_size (add_queue_message,
 488         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_SETUP,
 489         struct GNUNET_TRANSPORT_AddQueueMessage,
 490         NULL),
 491     //GNUNET_MQ_hd_fixed_size (del_queue_message,
 492     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_TEARDOWN,
 493     //                         struct GNUNET_TRANSPORT_DelQueueMessage,
 494     //                         NULL),
 495     //GNUNET_MQ_hd_fixed_size (send_message_ack,
 496     //                         GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG_ACK,
 497     //                         struct GNUNET_TRANSPORT_SendMessageToAck,
 498     //                         NULL),
 499   };
 500   struct GNUNET_SERVICE_Handle *h;
 501
 502   h = GNUNET_SERVICE_start ("transport",
 503                             tc_h->cfg,
 504                             &connect_cb,
 505                             &disconnect_cb,
 506                             tc_h,
 507                             mh);
 508   if (NULL == h)
 509     LOG (GNUNET_ERROR_TYPE_ERROR,
 510          "Failed starting service!\n");
 511   else
 512   {
 513     LOG (GNUNET_ERROR_TYPE_DEBUG,
 514         "Started service\n");
 515     /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_service, h);
 516   }
 517 }
 518
 519
 520 /**
 521  * @brief Task run at shutdown to kill communicator and clean up
 522  *
 523  * @param cls Closure - Process of communicator
 524  */
 525 static void
 526 shutdown_communicator (void *cls)
 527 {
 528   struct GNUNET_OS_Process *proc = cls;
 529
 530   if (GNUNET_OK != GNUNET_OS_process_kill (proc,
 531                                            SIGTERM))
 532   {
 533     LOG (GNUNET_ERROR_TYPE_WARNING,
 534         "Error shutting down communicator with SIGERM, trying SIGKILL\n");
 535     if (GNUNET_OK != GNUNET_OS_process_kill (proc,
 536                                              SIGKILL))
 537     {
 538       LOG (GNUNET_ERROR_TYPE_ERROR,
 539           "Error shutting down communicator with SIGERM and SIGKILL\n");
 540     }
 541   }
 542   GNUNET_OS_process_destroy (proc);
 543 }
 544
 545
 546 /**
 547  * @brief Start the communicator
 548  *
 549  * @param cfgname Name of the communicator
 550  */
 551 static void
 552 communicator_start (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h)
 553 {
 554   char *binary;
 555
 556   LOG (GNUNET_ERROR_TYPE_DEBUG,
 557       "communicator_start\n");
 558   binary = GNUNET_OS_get_libexec_binary_path ("gnunet-communicator-unix");
 559   tc_h->c_proc =
 560     GNUNET_OS_start_process (GNUNET_YES,
 561                              GNUNET_OS_INHERIT_STD_OUT_AND_ERR,
 562                              NULL, NULL, NULL,
 563                              binary,
 564                              "./gnunet-communicator-unix",
 565                              "-c",
 566                              tc_h->cfg_filename,
 567                              NULL);
 568   if (NULL == tc_h->c_proc)
 569   {
 570     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 571                 "Failed to start communicator!");
 572     return;
 573   }
 574   LOG (GNUNET_ERROR_TYPE_DEBUG,
 575       "started communicator\n");
 576   GNUNET_free (binary);
 577   /* TODO */ GNUNET_SCHEDULER_add_shutdown (&shutdown_communicator,
 578                                             tc_h->c_proc);
 579 }
 580
 581
 582 /**
 583  * @brief Start communicator part of transport service and communicator
 584  *
 585  * @param service_name Name of the service
 586  * @param cfg Configuration handle
 587  * @param communicator_available_cb Callback that is called when a new
 588  * @param add_address_cb Callback that is called when a new
 589  * communicator becomes available
 590  * @param cb_cls Closure to @a communicator_available_cb and @a
 591  *
 592  * @return Handle to the communicator duo
 593  */
 594 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *
 595 GNUNET_TRANSPORT_TESTING_transport_communicator_service_start
 596   (const char *service_name,
 597    const char *cfg_filename,
 598    GNUNET_TRANSPORT_TESTING_CommunicatorAvailableCallback communicator_available_cb,
 599    GNUNET_TRANSPORT_TESTING_AddAddressCallback add_address_cb,
 600    GNUNET_TRANSPORT_TESTING_QueueCreateReplyCallback queue_create_reply_cb,
 601    GNUNET_TRANSPORT_TESTING_AddQueueCallback add_queue_cb,
 602    void *cb_cls)
 603 {
 604   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h;
 605
 606   tc_h = GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle);
 607   tc_h->cfg_filename = GNUNET_strdup (cfg_filename);
 608   tc_h->cfg = GNUNET_CONFIGURATION_create ();
 609   if ( (GNUNET_SYSERR ==
 610         GNUNET_CONFIGURATION_load (tc_h->cfg,
 611                                    cfg_filename)) )
 612   {
 613     GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
 614                 _("Malformed configuration file `%s', exit ...\n"),
 615                   cfg_filename);
 616     return NULL;
 617   }
 618   tc_h->communicator_available_cb = communicator_available_cb;
 619   tc_h->add_address_cb = add_address_cb;
 620   tc_h->queue_create_reply_cb = queue_create_reply_cb;
 621   tc_h->add_queue_cb = add_queue_cb;
 622   tc_h->cb_cls = cb_cls;
 623
 624   /* Start communicator part of service */
 625   transport_communicator_start (tc_h);
 626
 627   /* Schedule start communicator */
 628   communicator_start (tc_h);
 629   return tc_h;
 630 }
 631
 632
 633 void
 634 GNUNET_TRANSPORT_TESTING_transport_communicator_open_queue
 635   (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorHandle *tc_h,
 636    const struct GNUNET_PeerIdentity *peer_id,
 637    const char *address)
 638 {
 639   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue;
 640   static uint32_t idgen;
 641   char *prefix;
 642   struct GNUNET_TRANSPORT_CreateQueue *msg;
 643   struct GNUNET_MQ_Envelope *env;
 644   size_t alen;
 645
 646   tc_queue = GNUNET_new (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue);
 647   prefix = GNUNET_HELLO_address_to_prefix (address);
 648   if (NULL == prefix)
 649   {
 650     GNUNET_break (0); /* We got an invalid address!? */
 651     return;
 652   }
 653   alen = strlen (address) + 1;
 654   env = GNUNET_MQ_msg_extra (msg,
 655                              alen,
 656                              GNUNET_MESSAGE_TYPE_TRANSPORT_QUEUE_CREATE);
 657   msg->request_id = htonl (idgen++);
 658   tc_queue->qid = msg->request_id;
 659   msg->receiver = *peer_id;
 660   tc_queue->peer_id = *peer_id;
 661   memcpy (&msg[1],
 662           address,
 663           alen);
 664   if (NULL != tc_h->c_mq)
 665   {
 666     GNUNET_MQ_send (tc_h->c_mq,
 667                     env);
 668   }
 669   else
 670   {
 671     tc_queue->open_queue_env = env;
 672   }
 673   GNUNET_CONTAINER_DLL_insert (tc_h->queue_head,
 674                                tc_h->queue_tail,
 675                                tc_queue);
 676 }
 677
 678
 679 struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *
 680 GNUNET_TRANSPORT_TESTING_transport_communicator_send
 681   (struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorQueue *tc_queue,
 682    const void *payload,
 683    size_t payload_size/*,
 684    GNUNET_TRANSPORT_TESTING_SuccessStatus cb,
 685    void *cb_cls*/)
 686 {
 687   struct GNUNET_TRANSPORT_TESTING_TransportCommunicatorTransmission *tc_t;
 688   struct GNUNET_TRANSPORT_SendMessageTo *msg;
 689   struct GNUNET_MQ_Envelope *env;
 690
 691   env = GNUNET_MQ_msg_extra (msg,
 692                              payload_size,
 693                              GNUNET_MESSAGE_TYPE_TRANSPORT_SEND_MSG);
 694   msg->qid = htonl (tc_queue->qid);
 695   msg->mid = tc_queue->mid++;
 696   msg->receiver = tc_queue->peer_id;
 697   memcpy (&msg[1],
 698           payload,
 699           payload_size);
 700   GNUNET_MQ_send (tc_queue->tc_h->c_mq,
 701                   env);
 702   return tc_t;
 703 }
 704