src/stream/test_stream_2peers_halfclose.c

   1 /*
   2      This file is part of GNUnet.
   3      (C) 2011, 2012 Christian Grothoff (and other contributing authors)
   4
   5      GNUnet is free software; you can redistribute it and/or modify
   6      it under the terms of the GNU General Public License as published
   7      by the Free Software Foundation; either version 3, or (at your
   8      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      General Public License for more details.
  14
  15      You should have received a copy of the GNU General Public License
  16      along with GNUnet; see the file COPYING.  If not, write to the
  17      Free Software Foundation, Inc., 59 Temple Place - Suite 330,
  18      Boston, MA 02111-1307, USA.
  19 */
  20
  21 /**
  22  * @file stream/test_stream_2peers_halfclose.c
  23  * @brief Testcases for Stream API halfclosed connections between 2 peers
  24  * @author Sree Harsha Totakura
  25  */
  26 #include "platform.h"
  27 #include "gnunet_util_lib.h"
  28 #include "gnunet_testbed_service.h"
  29 #include "gnunet_mesh_service.h"
  30 #include "gnunet_stream_lib.h"
  31
  32 /**
  33  * Number of peers
  34  */
  35 #define NUM_PEERS 2
  36
  37 #define TIME_REL_SECS(sec) \
  38   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, sec)
  39
  40 /**
  41  * Structure for holding peer's sockets and IO Handles
  42  */
  43 struct PeerData
  44 {
  45   /**
  46    * The testbed peer handle corresponding to this peer
  47    */
  48   struct GNUNET_TESTBED_Peer *peer;
  49
  50   /**
  51    * Peer's stream socket
  52    */
  53   struct GNUNET_STREAM_Socket *socket;
  54
  55   /**
  56    * Peer's io write handle
  57    */
  58   struct GNUNET_STREAM_WriteHandle *io_write_handle;
  59
  60   /**
  61    * Peer's io read handle
  62    */
  63   struct GNUNET_STREAM_ReadHandle *io_read_handle;
  64
  65   /**
  66    * Peer's shutdown handle
  67    */
  68   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
  69
  70   /**
  71    * Testbed operation handle specific for this peer
  72    */
  73   struct GNUNET_TESTBED_Operation *op;
  74
  75   /**
  76    * Our Peer id
  77    */
  78   struct GNUNET_PeerIdentity our_id;
  79
  80   /**
  81    * Bytes the peer has written
  82    */
  83   unsigned int bytes_wrote;
  84
  85   /**
  86    * Byte the peer has read
  87    */
  88   unsigned int bytes_read;
  89
  90   /**
  91    * GNUNET_YES if the peer has successfully completed the current test
  92    */
  93   unsigned int test_ok;
  94
  95   /**
  96    * The shutdown operation that has to be used by the stream_shutdown_task
  97    */
  98   int shutdown_operation;
  99 };
 100
 101
 102 /**
 103  * Enumeration for various tests that are to be passed in the same order as
 104  * below
 105  */
 106 enum Test
 107 {
 108   /**
 109    * Peer1 writing; Peer2 reading
 110    */
 111   PEER1_WRITE,
 112
 113   /**
 114    * Peer1 write shutdown; Peer2 should get an error when it tries to read;
 115    */
 116   PEER1_WRITE_SHUTDOWN,
 117
 118   /**
 119    * Peer1 reads; Peer2 writes (connection is halfclosed)
 120    */
 121   PEER1_HALFCLOSE_READ,
 122
 123   /**
 124    * Peer1 attempts to write; Should fail with stream already shutdown error
 125    */
 126   PEER1_HALFCLOSE_WRITE_FAIL,
 127
 128   /**
 129    * Peer1 read shutdown; Peer2 should get stream shutdown error during write
 130    */
 131   PEER1_READ_SHUTDOWN,
 132
 133   /**
 134    * All tests successfully finished
 135    */
 136   SUCCESS
 137 };
 138
 139
 140 /**
 141  * Different states in test setup
 142  */
 143 enum SetupState
 144 {
 145   /**
 146    * The initial state
 147    */
 148   INIT,
 149
 150   /**
 151    * Get the identity of peer 1
 152    */
 153   PEER1_GET_IDENTITY,
 154
 155   /**
 156    * Get the identity of peer 2
 157    */
 158   PEER2_GET_IDENTITY,
 159
 160   /**
 161    * Connect to stream service of peer 2
 162    */
 163   PEER2_STREAM_CONNECT,
 164
 165   /**
 166    * Connect to stream service of peer 1
 167    */
 168   PEER1_STREAM_CONNECT
 169
 170 };
 171
 172
 173 /**
 174  * Peer1 writes first and then calls for SHUT_WR
 175  * Peer2 reads first and then calls for SHUT_RD
 176  * Attempt to write again by Peer1 should be rejected
 177  * Attempt to read again by Peer2 should be rejected
 178  * Peer1 then reads from Peer2 which writes
 179  */
 180 static struct PeerData peer1;
 181 static struct PeerData peer2;
 182
 183 /**
 184  * Task for aborting the test case if it takes too long
 185  */
 186 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
 187
 188 /**
 189  * Task for reading from stream
 190  */
 191 static GNUNET_SCHEDULER_TaskIdentifier read_task;
 192
 193 static char *data = "ABCD";
 194
 195 /**
 196  * Handle to testbed operation
 197  */
 198 struct GNUNET_TESTBED_Operation *op;
 199
 200 /**
 201  * Final testing result
 202  */
 203 static int result;
 204
 205 /**
 206  * Current running test
 207  */
 208 enum Test current_test;
 209
 210 /**
 211  * State is test setup
 212  */
 213 enum SetupState setup_state;
 214
 215
 216 /**
 217  * Input processor
 218  *
 219  * @param cls the closure from GNUNET_STREAM_write/read
 220  * @param status the status of the stream at the time this function is called
 221  * @param data traffic from the other side
 222  * @param size the number of bytes available in data read
 223  * @return number of bytes of processed from 'data' (any data remaining should be
 224  *         given to the next time the read processor is called).
 225  */
 226 static size_t
 227 input_processor (void *cls,
 228                  enum GNUNET_STREAM_Status status,
 229                  const void *input_data,
 230                  size_t size);
 231
 232
 233 /**
 234  * The transition function; responsible for the transitions among tests
 235  */
 236 static void
 237 transition();
 238
 239
 240 /**
 241  * Task for calling STREAM_read
 242  *
 243  * @param cls the peer data entity
 244  * @param tc the task context
 245  */
 246 static void
 247 stream_read_task (void *cls,
 248                   const struct GNUNET_SCHEDULER_TaskContext *tc)
 249 {
 250   struct PeerData *peer = cls;
 251
 252   peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
 253                                              GNUNET_TIME_relative_multiply
 254                                              (GNUNET_TIME_UNIT_SECONDS, 5),
 255                                              &input_processor,
 256                                              cls);
 257   switch (current_test)
 258     {
 259     case PEER1_WRITE_SHUTDOWN:
 260       GNUNET_assert (&peer2 == peer);
 261       GNUNET_assert (NULL == peer->io_read_handle);
 262       peer2.test_ok = GNUNET_YES;
 263       transition ();            /* to PEER1_HALFCLOSE_READ */
 264       break;
 265     default:
 266       GNUNET_assert (NULL != peer->io_read_handle);
 267     }
 268 }
 269
 270
 271 /**
 272  * The write completion function; called upon writing some data to stream or
 273  * upon error
 274  *
 275  * @param cls the closure from GNUNET_STREAM_write/read
 276  * @param status the status of the stream at the time this function is called
 277  * @param size the number of bytes read or written
 278  */
 279 static void
 280 write_completion (void *cls,
 281                   enum GNUNET_STREAM_Status status,
 282                   size_t size);
 283
 284
 285 /**
 286  * Task for calling STREAM_write
 287  *
 288  * @param cls the peer data entity
 289  * @param tc the task context
 290  */
 291 static void
 292 stream_write_task (void *cls,
 293                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 294 {
 295   struct PeerData *peer = cls;
 296
 297   peer->io_write_handle =
 298     GNUNET_STREAM_write (peer->socket,
 299                          (void *) data,
 300                          strlen(data) - peer->bytes_wrote,
 301                          GNUNET_TIME_relative_multiply
 302                          (GNUNET_TIME_UNIT_SECONDS, 5),
 303                          &write_completion,
 304                          peer);
 305   switch (current_test)
 306     {
 307     case PEER1_HALFCLOSE_WRITE_FAIL:
 308       GNUNET_assert (&peer1 == peer);
 309       GNUNET_assert (NULL == peer->io_write_handle);
 310       transition();             /* To PEER1_READ_SHUTDOWN */
 311       break;
 312     case PEER1_READ_SHUTDOWN:
 313       GNUNET_assert (&peer2 == peer);
 314       GNUNET_assert (NULL == peer->io_write_handle);
 315       transition ();            /* To SUCCESS */
 316       break;
 317     default:
 318         GNUNET_assert (NULL != peer->io_write_handle);
 319     }
 320 }
 321
 322
 323 /**
 324  * Close sockets and stop testing deamons nicely
 325  */
 326 static void
 327 do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 328 {
 329   if (NULL != peer2.socket)
 330     GNUNET_STREAM_close (peer2.socket);
 331   if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 332     GNUNET_SCHEDULER_cancel (abort_task);
 333   if (NULL != peer2.op)
 334     GNUNET_TESTBED_operation_done (peer2.op);
 335   else
 336     GNUNET_SCHEDULER_shutdown (); /* For shutting down testbed */
 337 }
 338
 339
 340 /**
 341  * Completion callback for shutdown
 342  *
 343  * @param cls the closure from GNUNET_STREAM_shutdown call
 344  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 345  *          SHUT_RDWR)
 346  */
 347 void
 348 shutdown_completion (void *cls,
 349                      int operation)
 350 {
 351   switch (current_test)
 352     {
 353     case PEER1_WRITE:
 354       GNUNET_assert (0);
 355     case PEER1_WRITE_SHUTDOWN:
 356       GNUNET_assert (cls == &peer1);
 357       GNUNET_assert (SHUT_WR == operation);
 358       peer1.test_ok = GNUNET_YES;
 359       /* Peer2 should read with error */
 360       peer2.bytes_read = 0;
 361       GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 362       break;
 363     case PEER1_READ_SHUTDOWN:
 364       peer1.test_ok = GNUNET_YES;
 365       peer2.bytes_wrote = 0;
 366       GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 367       break;
 368     case PEER1_HALFCLOSE_READ:
 369     case PEER1_HALFCLOSE_WRITE_FAIL:
 370     case SUCCESS:
 371       GNUNET_assert (0);        /* We shouldn't reach here */
 372     }
 373 }
 374
 375
 376 /**
 377  * Task for calling STREAM_shutdown
 378  *
 379  * @param cls the peer entity
 380  * @param tc the TaskContext
 381  */
 382 static void
 383 stream_shutdown_task (void *cls,
 384                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 385 {
 386   struct PeerData *peer = cls;
 387
 388   peer->shutdown_handle = GNUNET_STREAM_shutdown (peer->socket,
 389                                                   peer->shutdown_operation,
 390                                                   &shutdown_completion,
 391                                                   peer);
 392   GNUNET_assert (NULL != peer->shutdown_handle);
 393 }
 394
 395
 396 /**
 397  * Something went wrong and timed out. Kill everything and set error flag
 398  */
 399 static void
 400 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 401 {
 402   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 403   if (0 != read_task)
 404     {
 405       GNUNET_SCHEDULER_cancel (read_task);
 406     }
 407   result = GNUNET_SYSERR;
 408   abort_task = 0;
 409   do_close (cls, tc);
 410 }
 411
 412
 413 /**
 414  * The transition function; responsible for the transitions among tests
 415  */
 416 static void
 417 transition()
 418 {
 419   if ((GNUNET_YES == peer1.test_ok) && (GNUNET_YES == peer2.test_ok))
 420     {
 421       peer1.test_ok = GNUNET_NO;
 422       peer2.test_ok = GNUNET_NO;
 423       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 424                   "TEST %d SUCCESSFULL\n", current_test);
 425       switch (current_test)
 426         {
 427         case PEER1_WRITE:
 428           current_test = PEER1_WRITE_SHUTDOWN;
 429           /* Peer1 should shutdown writing */
 430           peer1.shutdown_operation = SHUT_WR;
 431           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 432           break;
 433         case PEER1_WRITE_SHUTDOWN:
 434           current_test = PEER1_HALFCLOSE_READ;
 435           /* Peer2 should be able to write successfully */
 436           peer2.bytes_wrote = 0;
 437           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 438
 439           /* Peer1 should be able to read successfully */
 440           peer1.bytes_read = 0;
 441           GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
 442           break;
 443         case PEER1_HALFCLOSE_READ:
 444           current_test = PEER1_HALFCLOSE_WRITE_FAIL;
 445           peer1.bytes_wrote = 0;
 446           peer2.bytes_read = 0;
 447           peer2.test_ok = GNUNET_YES;
 448           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer1);
 449           break;
 450         case PEER1_HALFCLOSE_WRITE_FAIL:
 451           current_test = PEER1_READ_SHUTDOWN;
 452           peer1.shutdown_operation = SHUT_RD;
 453           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 454           break;
 455         case PEER1_READ_SHUTDOWN:
 456           current_test = SUCCESS;
 457           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 458                       "All tests successful\n");
 459           GNUNET_SCHEDULER_add_now (&do_close, NULL);
 460           break;
 461         case SUCCESS:
 462           GNUNET_assert (0);    /* We shouldn't reach here */
 463
 464         }
 465     }
 466 }
 467
 468 /**
 469  * The write completion function; called upon writing some data to stream or
 470  * upon error
 471  *
 472  * @param cls the closure from GNUNET_STREAM_write/read
 473  * @param status the status of the stream at the time this function is called
 474  * @param size the number of bytes read or written
 475  */
 476 static void
 477 write_completion (void *cls,
 478                   enum GNUNET_STREAM_Status status,
 479                   size_t size)
 480 {
 481   struct PeerData *peer = cls;
 482
 483   switch (current_test)
 484     {
 485     case PEER1_WRITE:
 486     case PEER1_HALFCLOSE_READ:
 487
 488     GNUNET_assert (GNUNET_STREAM_OK == status);
 489     GNUNET_assert (size <= strlen (data));
 490     peer->bytes_wrote += size;
 491
 492     if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 493       {
 494         GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 495       }
 496     else
 497       {
 498         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 499                     "Writing completed\n");
 500
 501         if (&peer1 == peer)
 502           {
 503             peer1.test_ok = GNUNET_YES;
 504             transition ();       /* to PEER1_WRITE_SHUTDOWN */
 505           }
 506         else            /* This will happen during PEER1_HALFCLOSE_READ */
 507           {
 508             peer2.test_ok = GNUNET_YES;
 509             transition ();      /* to PEER1_HALFCLOSE_WRITE_FAIL */
 510           }
 511       }
 512     break;
 513     case PEER1_HALFCLOSE_WRITE_FAIL:
 514       GNUNET_assert (peer == &peer1);
 515       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 516       GNUNET_assert (0 == size);
 517       peer1.test_ok = GNUNET_YES;
 518       break;
 519     case PEER1_READ_SHUTDOWN:
 520       GNUNET_assert (peer == &peer2);
 521       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 522       GNUNET_assert (0 == size);
 523       peer2.test_ok = GNUNET_YES;
 524       break;
 525     case PEER1_WRITE_SHUTDOWN:
 526     case SUCCESS:
 527       GNUNET_assert (0);        /* We shouldn't reach here */
 528     }
 529 }
 530
 531
 532 /**
 533  * Function executed after stream has been established
 534  *
 535  * @param cls the closure from GNUNET_STREAM_open
 536  * @param socket socket to use to communicate with the other side (read/write)
 537  */
 538 static void
 539 stream_open_cb (void *cls,
 540                 struct GNUNET_STREAM_Socket *socket)
 541 {
 542   struct PeerData *peer;
 543
 544   GNUNET_assert (socket == peer1.socket);
 545   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 546               "%s: Stream established from peer1\n",
 547               GNUNET_i2s (&peer1.our_id));
 548   peer = (struct PeerData *) cls;
 549   peer->bytes_wrote = 0;
 550   GNUNET_assert (socket == peer1.socket);
 551   GNUNET_assert (socket == peer->socket);
 552   peer1.test_ok = GNUNET_NO;
 553   peer2.test_ok = GNUNET_NO;
 554   current_test = PEER1_WRITE;
 555   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 556 }
 557
 558
 559 /**
 560  * Input processor
 561  *
 562  * @param cls the closure from GNUNET_STREAM_write/read
 563  * @param status the status of the stream at the time this function is called
 564  * @param data traffic from the other side
 565  * @param size the number of bytes available in data read
 566  * @return number of bytes of processed from 'data' (any data remaining should be
 567  *         given to the next time the read processor is called).
 568  */
 569 static size_t
 570 input_processor (void *cls,
 571                  enum GNUNET_STREAM_Status status,
 572                  const void *input_data,
 573                  size_t size)
 574 {
 575   struct PeerData *peer;
 576
 577   peer = (struct PeerData *) cls;
 578
 579   switch (current_test)
 580     {
 581     case PEER1_WRITE:
 582     case PEER1_HALFCLOSE_READ:
 583       if (GNUNET_STREAM_TIMEOUT == status)
 584         {
 585           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 586                       "Read operation timedout - reading again!\n");
 587           GNUNET_assert (0 == size);
 588           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 589           return 0;
 590         }
 591
 592       GNUNET_assert (GNUNET_STREAM_OK == status);
 593       GNUNET_assert (size <= strlen (data));
 594       GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
 595                                    (const char *) input_data,
 596                                    size));
 597       peer->bytes_read += size;
 598
 599       if (peer->bytes_read < strlen (data))
 600         {
 601           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 602         }
 603       else
 604         {
 605           if (&peer2 == peer) /* Peer2 has completed reading; should write */
 606             {
 607               peer2.test_ok = GNUNET_YES;
 608               transition ();    /* Transition to PEER1_WRITE_SHUTDOWN */
 609             }
 610           else         /* Peer1 has completed reading. End of tests */
 611             {
 612               peer1.test_ok = GNUNET_YES;
 613               transition ();    /* to PEER1_HALFCLOSE_WRITE_FAIL */
 614             }
 615         }
 616       break;
 617     case PEER1_WRITE_SHUTDOWN:
 618       GNUNET_assert (0);        /* This callback will not be called when stream
 619                                    is shutdown */
 620       break;
 621     case PEER1_HALFCLOSE_WRITE_FAIL:
 622     case PEER1_READ_SHUTDOWN:
 623     case SUCCESS:
 624       GNUNET_assert (0);        /* We shouldn't reach here */
 625     }
 626
 627   return size;
 628 }
 629
 630
 631 /**
 632  * Scheduler call back; to be executed when a new stream is connected
 633  * Called from listen connect for peer2
 634  */
 635 static void
 636 stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 637 {
 638   read_task = GNUNET_SCHEDULER_NO_TASK;
 639   GNUNET_assert (NULL != cls);
 640   peer2.bytes_read = 0;
 641   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 642 }
 643
 644
 645 /**
 646  * Functions of this type are called upon new stream connection from other peers
 647  *
 648  * @param cls the closure from GNUNET_STREAM_listen
 649  * @param socket the socket representing the stream
 650  * @param initiator the identity of the peer who wants to establish a stream
 651  *            with us
 652  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 653  *             stream (the socket will be invalid after the call)
 654  */
 655 static int
 656 stream_listen_cb (void *cls,
 657                   struct GNUNET_STREAM_Socket *socket,
 658                   const struct GNUNET_PeerIdentity *initiator)
 659 {
 660   if ((NULL == socket) || (NULL == initiator))
 661   {
 662     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "Binding error\n");
 663     if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 664       GNUNET_SCHEDULER_cancel (abort_task);
 665     abort_task = GNUNET_SCHEDULER_add_now (&do_abort, NULL);
 666     return GNUNET_OK;
 667   }
 668   GNUNET_assert (socket != peer1.socket);
 669   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 670               "%s: Peer connected: %s\n",
 671               GNUNET_i2s (&peer2.our_id),
 672               GNUNET_i2s(initiator));
 673   peer2.socket = socket;
 674   /* FIXME: reading should be done right now instead of a scheduled call */
 675   read_task = GNUNET_SCHEDULER_add_now (&stream_read, (void *) socket);
 676   return GNUNET_OK;
 677 }
 678
 679
 680 /**
 681  * Listen success callback; connects a peer to stream as client
 682  */
 683 static void
 684 stream_connect (void);
 685
 686
 687 /**
 688  * Adapter function called to destroy a connection to
 689  * a service.
 690  *
 691  * @param cls closure
 692  * @param op_result service handle returned from the connect adapter
 693  */
 694 static void
 695 stream_da (void *cls, void *op_result)
 696 {
 697   struct GNUNET_STREAM_ListenSocket *lsocket;
 698
 699   if (&peer2 == cls)
 700   {
 701     lsocket = op_result;
 702     GNUNET_STREAM_listen_close (lsocket);
 703     if (NULL != peer1.op)
 704       GNUNET_TESTBED_operation_done (peer1.op);
 705     else
 706       GNUNET_SCHEDULER_shutdown ();
 707     return;
 708   }
 709   if (&peer1 == cls)
 710   {
 711     GNUNET_assert (op_result == peer1.socket);
 712     GNUNET_STREAM_close (peer1.socket);
 713     GNUNET_SCHEDULER_shutdown (); /* Exit point of the test */
 714     return;
 715   }
 716   GNUNET_assert (0);
 717 }
 718
 719
 720 /**
 721  * Adapter function called to establish a connection to
 722  * a service.
 723  *
 724  * @param cls closure
 725  * @param cfg configuration of the peer to connect to; will be available until
 726  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 727  *          from GNUNET_TESTBED_service_connect()
 728  * @return service handle to return in 'op_result', NULL on error
 729  */
 730 static void *
 731 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 732 {
 733   struct GNUNET_STREAM_ListenSocket *lsocket;
 734
 735   switch (setup_state)
 736   {
 737   case PEER2_STREAM_CONNECT:
 738     lsocket = GNUNET_STREAM_listen (cfg, 10, &stream_listen_cb, NULL,
 739                                     GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 740                                     &stream_connect, GNUNET_STREAM_OPTION_END);
 741     GNUNET_assert (NULL != lsocket);
 742     return lsocket;
 743   case PEER1_STREAM_CONNECT:
 744     peer1.socket = GNUNET_STREAM_open (cfg, &peer2.our_id, 10, &stream_open_cb,
 745                                        &peer1, GNUNET_STREAM_OPTION_END);
 746     GNUNET_assert (NULL != peer1.socket);
 747     return peer1.socket;
 748   default:
 749     GNUNET_assert (0);
 750   }
 751 }
 752
 753
 754 /**
 755  * Listen success callback; connects a peer to stream as client
 756  */
 757 static void
 758 stream_connect (void)
 759 {
 760   GNUNET_assert (PEER2_STREAM_CONNECT == setup_state);
 761   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream listen open successful\n");
 762   peer1.op = GNUNET_TESTBED_service_connect (&peer1, peer1.peer, "stream",
 763                                              NULL, NULL,
 764                                              stream_ca, stream_da, &peer1);
 765   setup_state = PEER1_STREAM_CONNECT;
 766 }
 767
 768
 769 /**
 770  * Callback to be called when the requested peer information is available
 771  *
 772  * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
 773  * @param op the operation this callback corresponds to
 774  * @param pinfo the result; will be NULL if the operation has failed
 775  * @param emsg error message if the operation has failed; will be NULL if the
 776  *          operation is successfull
 777  */
 778 static void
 779 peerinfo_cb (void *cb_cls, struct GNUNET_TESTBED_Operation *op_,
 780              const struct GNUNET_TESTBED_PeerInformation *pinfo,
 781              const char *emsg)
 782 {
 783   GNUNET_assert (NULL == emsg);
 784   GNUNET_assert (op == op_);
 785   switch (setup_state)
 786   {
 787   case PEER1_GET_IDENTITY:
 788     memcpy (&peer1.our_id, pinfo->result.id,
 789             sizeof (struct GNUNET_PeerIdentity));
 790     GNUNET_TESTBED_operation_done (op);
 791     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 1 id: %s\n", GNUNET_i2s
 792                 (&peer1.our_id));
 793     op = GNUNET_TESTBED_peer_get_information (peer2.peer,
 794                                               GNUNET_TESTBED_PIT_IDENTITY,
 795                                               &peerinfo_cb, NULL);
 796     setup_state = PEER2_GET_IDENTITY;
 797     break;
 798   case PEER2_GET_IDENTITY:
 799     memcpy (&peer2.our_id, pinfo->result.id,
 800             sizeof (struct GNUNET_PeerIdentity));
 801     GNUNET_TESTBED_operation_done (op);
 802     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 2 id: %s\n", GNUNET_i2s
 803                 (&peer2.our_id));
 804     peer2.op = GNUNET_TESTBED_service_connect (&peer2, peer2.peer, "stream",
 805                                                NULL, NULL,
 806                                                stream_ca, stream_da, &peer2);
 807     setup_state = PEER2_STREAM_CONNECT;
 808     break;
 809   default:
 810     GNUNET_assert (0);
 811   }
 812 }
 813
 814
 815 /**
 816  * Controller event callback
 817  *
 818  * @param cls NULL
 819  * @param event the controller event
 820  */
 821 static void
 822 controller_event_cb (void *cls,
 823                      const struct GNUNET_TESTBED_EventInformation *event)
 824 {
 825   switch (event->type)
 826   {
 827   case GNUNET_TESTBED_ET_CONNECT:
 828     GNUNET_assert (INIT == setup_state);
 829     GNUNET_TESTBED_operation_done (op);
 830     op = GNUNET_TESTBED_peer_get_information (peer1.peer,
 831                                               GNUNET_TESTBED_PIT_IDENTITY,
 832                                               &peerinfo_cb, NULL);
 833     setup_state = PEER1_GET_IDENTITY;
 834     break;
 835   case GNUNET_TESTBED_ET_OPERATION_FINISHED:
 836     switch (setup_state)
 837     {
 838     case PEER1_STREAM_CONNECT:
 839     case PEER2_STREAM_CONNECT:
 840       GNUNET_assert (NULL == event->details.operation_finished.emsg);
 841       break;
 842     default:
 843       GNUNET_assert (0);
 844     }
 845     break;
 846   default:
 847     GNUNET_assert (0);
 848   }
 849 }
 850
 851
 852 /**
 853  * Signature of a main function for a testcase.
 854  *
 855  * @param cls closure
 856  * @param num_peers number of peers in 'peers'
 857  * @param peers handle to peers run in the testbed
 858  */
 859 static void
 860 test_master (void *cls, unsigned int num_peers,
 861              struct GNUNET_TESTBED_Peer **peers)
 862 {
 863   GNUNET_assert (NULL != peers);
 864   GNUNET_assert (NULL != peers[0]);
 865   GNUNET_assert (NULL != peers[1]);
 866   peer1.peer = peers[0];
 867   peer2.peer = peers[1];
 868   op = GNUNET_TESTBED_overlay_connect (NULL, NULL, NULL, peer2.peer, peer1.peer);
 869   setup_state = INIT;
 870   abort_task =
 871     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 872                                   (GNUNET_TIME_UNIT_SECONDS, 1000), &do_abort,
 873                                   NULL);
 874 }
 875
 876
 877 /**
 878  * Main function
 879  */
 880 int main (int argc, char **argv)
 881 {
 882   uint64_t event_mask;
 883
 884   result = GNUNET_NO;
 885   event_mask = 0;
 886   event_mask |= (1LL << GNUNET_TESTBED_ET_CONNECT);
 887   event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
 888   (void) GNUNET_TESTBED_test_run ("test_stream_2peers_halfclose",
 889                                   "test_stream_local.conf", NUM_PEERS,
 890                                   event_mask,
 891                                   &controller_event_cb, NULL, &test_master,
 892                                   NULL);
 893   if (GNUNET_SYSERR == result)
 894     return 1;
 895   return 0;
 896 }