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
  27 #include <string.h>
  28
  29 #include "platform.h"
  30 #include "gnunet_util_lib.h"
  31 #include "gnunet_testbed_service.h"
  32 #include "gnunet_mesh_service.h"
  33 #include "gnunet_stream_lib.h"
  34
  35 #define VERBOSE 1
  36
  37 /**
  38  * Number of peers
  39  */
  40 #define NUM_PEERS 2
  41
  42 #define TIME_REL_SECS(sec) \
  43   GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, sec)
  44
  45 /**
  46  * Structure for holding peer's sockets and IO Handles
  47  */
  48 struct PeerData
  49 {
  50   /**
  51    * The testbed peer handle corresponding to this peer
  52    */
  53   struct GNUNET_TESTBED_Peer *peer;
  54
  55   /**
  56    * Peer's stream socket
  57    */
  58   struct GNUNET_STREAM_Socket *socket;
  59
  60   /**
  61    * Peer's io write handle
  62    */
  63   struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
  64
  65   /**
  66    * Peer's io read handle
  67    */
  68   struct GNUNET_STREAM_IOReadHandle *io_read_handle;
  69
  70   /**
  71    * Peer's shutdown handle
  72    */
  73   struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
  74
  75   /**
  76    * Testbed operation handle specific for this peer
  77    */
  78   struct GNUNET_TESTBED_Operation *op;
  79
  80   /**
  81    * Our Peer id
  82    */
  83   struct GNUNET_PeerIdentity our_id;
  84
  85   /**
  86    * Bytes the peer has written
  87    */
  88   unsigned int bytes_wrote;
  89
  90   /**
  91    * Byte the peer has read
  92    */
  93   unsigned int bytes_read;
  94
  95   /**
  96    * GNUNET_YES if the peer has successfully completed the current test
  97    */
  98   unsigned int test_ok;
  99
 100   /**
 101    * The shutdown operation that has to be used by the stream_shutdown_task
 102    */
 103   int shutdown_operation;
 104 };
 105
 106
 107 /**
 108  * Enumeration for various tests that are to be passed in the same order as
 109  * below
 110  */
 111 enum Test
 112 {
 113   /**
 114    * Peer1 writing; Peer2 reading
 115    */
 116   PEER1_WRITE,
 117
 118   /**
 119    * Peer1 write shutdown; Peer2 should get an error when it tries to read;
 120    */
 121   PEER1_WRITE_SHUTDOWN,
 122
 123   /**
 124    * Peer1 reads; Peer2 writes (connection is halfclosed)
 125    */
 126   PEER1_HALFCLOSE_READ,
 127
 128   /**
 129    * Peer1 attempts to write; Should fail with stream already shutdown error
 130    */
 131   PEER1_HALFCLOSE_WRITE_FAIL,
 132
 133   /**
 134    * Peer1 read shutdown; Peer2 should get stream shutdown error during write
 135    */
 136   PEER1_READ_SHUTDOWN,
 137
 138   /**
 139    * All tests successfully finished
 140    */
 141   SUCCESS
 142 };
 143
 144
 145 /**
 146  * Different states in test setup
 147  */
 148 enum SetupState
 149 {
 150   /**
 151    * The initial state
 152    */
 153   INIT,
 154
 155   /**
 156    * Get the identity of peer 1
 157    */
 158   PEER1_GET_IDENTITY,
 159
 160   /**
 161    * Get the identity of peer 2
 162    */
 163   PEER2_GET_IDENTITY,
 164
 165   /**
 166    * Connect to stream service of peer 2
 167    */
 168   PEER2_STREAM_CONNECT,
 169
 170   /**
 171    * Connect to stream service of peer 1
 172    */
 173   PEER1_STREAM_CONNECT
 174
 175 };
 176
 177
 178 /**
 179  * Peer1 writes first and then calls for SHUT_WR
 180  * Peer2 reads first and then calls for SHUT_RD
 181  * Attempt to write again by Peer1 should be rejected
 182  * Attempt to read again by Peer2 should be rejected
 183  * Peer1 then reads from Peer2 which writes
 184  */
 185 static struct PeerData peer1;
 186 static struct PeerData peer2;
 187
 188 /**
 189  * Task for aborting the test case if it takes too long
 190  */
 191 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
 192
 193 /**
 194  * Task for reading from stream
 195  */
 196 static GNUNET_SCHEDULER_TaskIdentifier read_task;
 197
 198 static char *data = "ABCD";
 199
 200 /**
 201  * Handle to testbed operation
 202  */
 203 struct GNUNET_TESTBED_Operation *op;
 204
 205 /**
 206  * Final testing result
 207  */
 208 static int result;
 209
 210 /**
 211  * Current running test
 212  */
 213 enum Test current_test;
 214
 215 /**
 216  * State is test setup
 217  */
 218 enum SetupState setup_state;
 219
 220
 221 /**
 222  * Input processor
 223  *
 224  * @param cls the closure from GNUNET_STREAM_write/read
 225  * @param status the status of the stream at the time this function is called
 226  * @param data traffic from the other side
 227  * @param size the number of bytes available in data read
 228  * @return number of bytes of processed from 'data' (any data remaining should be
 229  *         given to the next time the read processor is called).
 230  */
 231 static size_t
 232 input_processor (void *cls,
 233                  enum GNUNET_STREAM_Status status,
 234                  const void *input_data,
 235                  size_t size);
 236
 237
 238 /**
 239  * The transition function; responsible for the transitions among tests
 240  */
 241 static void
 242 transition();
 243
 244
 245 /**
 246  * Task for calling STREAM_read
 247  *
 248  * @param cls the peer data entity
 249  * @param tc the task context
 250  */
 251 static void
 252 stream_read_task (void *cls,
 253                   const struct GNUNET_SCHEDULER_TaskContext *tc)
 254 {
 255   struct PeerData *peer = cls;
 256
 257   peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
 258                                              GNUNET_TIME_relative_multiply
 259                                              (GNUNET_TIME_UNIT_SECONDS, 5),
 260                                              &input_processor,
 261                                              cls);
 262   switch (current_test)
 263     {
 264     case PEER1_WRITE_SHUTDOWN:
 265       GNUNET_assert (&peer2 == peer);
 266       GNUNET_assert (NULL == peer->io_read_handle);
 267       transition ();            /* to PEER1_HALFCLOSE_READ */
 268       break;
 269     default:
 270       GNUNET_assert (NULL != peer->io_read_handle);
 271     }
 272 }
 273
 274
 275 /**
 276  * The write completion function; called upon writing some data to stream or
 277  * upon error
 278  *
 279  * @param cls the closure from GNUNET_STREAM_write/read
 280  * @param status the status of the stream at the time this function is called
 281  * @param size the number of bytes read or written
 282  */
 283 static void
 284 write_completion (void *cls,
 285                   enum GNUNET_STREAM_Status status,
 286                   size_t size);
 287
 288
 289 /**
 290  * Task for calling STREAM_write
 291  *
 292  * @param cls the peer data entity
 293  * @param tc the task context
 294  */
 295 static void
 296 stream_write_task (void *cls,
 297                    const struct GNUNET_SCHEDULER_TaskContext *tc)
 298 {
 299   struct PeerData *peer = cls;
 300
 301   peer->io_write_handle =
 302     GNUNET_STREAM_write (peer->socket,
 303                          (void *) data,
 304                          strlen(data) - peer->bytes_wrote,
 305                          GNUNET_TIME_relative_multiply
 306                          (GNUNET_TIME_UNIT_SECONDS, 5),
 307                          &write_completion,
 308                          peer);
 309   switch (current_test)
 310     {
 311     case PEER1_HALFCLOSE_WRITE_FAIL:
 312       GNUNET_assert (&peer1 == peer);
 313       GNUNET_assert (NULL == peer->io_write_handle);
 314       transition();             /* To PEER1_READ_SHUTDOWN */
 315       break;
 316     case PEER1_READ_SHUTDOWN:
 317       GNUNET_assert (&peer2 == peer);
 318       GNUNET_assert (NULL == peer->io_write_handle);
 319       transition ();            /* To SUCCESS */
 320       break;
 321     default:
 322         GNUNET_assert (NULL != peer->io_write_handle);
 323     }
 324 }
 325
 326
 327 /**
 328  * Close sockets and stop testing deamons nicely
 329  */
 330 static void
 331 do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 332 {
 333   if (NULL != peer2.socket)
 334     GNUNET_STREAM_close (peer2.socket);
 335   if (GNUNET_SCHEDULER_NO_TASK != abort_task)
 336     GNUNET_SCHEDULER_cancel (abort_task);
 337   if (NULL != peer2.op)
 338     GNUNET_TESTBED_operation_done (peer2.op);
 339   else
 340     GNUNET_SCHEDULER_shutdown (); /* For shutting down testbed */
 341 }
 342
 343
 344 /**
 345  * Completion callback for shutdown
 346  *
 347  * @param cls the closure from GNUNET_STREAM_shutdown call
 348  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 349  *          SHUT_RDWR)
 350  */
 351 void
 352 shutdown_completion (void *cls,
 353                      int operation)
 354 {
 355   switch (current_test)
 356     {
 357     case PEER1_WRITE:
 358       GNUNET_assert (0);
 359     case PEER1_WRITE_SHUTDOWN:
 360       GNUNET_assert (cls == &peer1);
 361       GNUNET_assert (SHUT_WR == operation);
 362       peer1.test_ok = GNUNET_YES;
 363       /* Peer2 should read with error */
 364       peer2.bytes_read = 0;
 365       GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 366       break;
 367     case PEER1_READ_SHUTDOWN:
 368       peer1.test_ok = GNUNET_YES;
 369       peer2.bytes_wrote = 0;
 370       GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 371       break;
 372     case PEER1_HALFCLOSE_READ:
 373     case PEER1_HALFCLOSE_WRITE_FAIL:
 374     case SUCCESS:
 375       GNUNET_assert (0);        /* We shouldn't reach here */
 376     }
 377 }
 378
 379
 380 /**
 381  * Task for calling STREAM_shutdown
 382  *
 383  * @param cls the peer entity
 384  * @param tc the TaskContext
 385  */
 386 static void
 387 stream_shutdown_task (void *cls,
 388                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 389 {
 390   struct PeerData *peer = cls;
 391
 392   peer->shutdown_handle = GNUNET_STREAM_shutdown (peer->socket,
 393                                                   peer->shutdown_operation,
 394                                                   &shutdown_completion,
 395                                                   peer);
 396   GNUNET_assert (NULL != peer->shutdown_handle);
 397 }
 398
 399
 400 /**
 401  * Something went wrong and timed out. Kill everything and set error flag
 402  */
 403 static void
 404 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 405 {
 406   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 407   if (0 != read_task)
 408     {
 409       GNUNET_SCHEDULER_cancel (read_task);
 410     }
 411   result = GNUNET_SYSERR;
 412   abort_task = 0;
 413   do_close (cls, tc);
 414 }
 415
 416
 417 /**
 418  * The transition function; responsible for the transitions among tests
 419  */
 420 static void
 421 transition()
 422 {
 423   if ((GNUNET_YES == peer1.test_ok) && (GNUNET_YES == peer2.test_ok))
 424     {
 425       peer1.test_ok = GNUNET_NO;
 426       peer2.test_ok = GNUNET_NO;
 427       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 428                   "TEST %d SUCCESSFULL\n", current_test);
 429       switch (current_test)
 430         {
 431         case PEER1_WRITE:
 432           current_test = PEER1_WRITE_SHUTDOWN;
 433           /* Peer1 should shutdown writing */
 434           peer1.shutdown_operation = SHUT_WR;
 435           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 436           break;
 437         case PEER1_WRITE_SHUTDOWN:
 438           current_test = PEER1_HALFCLOSE_READ;
 439           /* Peer2 should be able to write successfully */
 440           peer2.bytes_wrote = 0;
 441           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 442
 443           /* Peer1 should be able to read successfully */
 444           peer1.bytes_read = 0;
 445           GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
 446           break;
 447         case PEER1_HALFCLOSE_READ:
 448           current_test = PEER1_HALFCLOSE_WRITE_FAIL;
 449           peer1.bytes_wrote = 0;
 450           peer2.bytes_read = 0;
 451           peer2.test_ok = GNUNET_YES;
 452           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer1);
 453           break;
 454         case PEER1_HALFCLOSE_WRITE_FAIL:
 455           current_test = PEER1_READ_SHUTDOWN;
 456           peer1.shutdown_operation = SHUT_RD;
 457           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 458           break;
 459         case PEER1_READ_SHUTDOWN:
 460           current_test = SUCCESS;
 461           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 462                       "All tests successful\n");
 463           GNUNET_SCHEDULER_add_now (&do_close, NULL);
 464           break;
 465         case SUCCESS:
 466           GNUNET_assert (0);    /* We shouldn't reach here */
 467
 468         }
 469     }
 470 }
 471
 472 /**
 473  * The write completion function; called upon writing some data to stream or
 474  * upon error
 475  *
 476  * @param cls the closure from GNUNET_STREAM_write/read
 477  * @param status the status of the stream at the time this function is called
 478  * @param size the number of bytes read or written
 479  */
 480 static void
 481 write_completion (void *cls,
 482                   enum GNUNET_STREAM_Status status,
 483                   size_t size)
 484 {
 485   struct PeerData *peer = cls;
 486
 487   switch (current_test)
 488     {
 489     case PEER1_WRITE:
 490     case PEER1_HALFCLOSE_READ:
 491
 492     GNUNET_assert (GNUNET_STREAM_OK == status);
 493     GNUNET_assert (size <= strlen (data));
 494     peer->bytes_wrote += size;
 495
 496     if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 497       {
 498         GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 499       }
 500     else
 501       {
 502         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 503                     "Writing completed\n");
 504
 505         if (&peer1 == peer)
 506           {
 507             peer1.test_ok = GNUNET_YES;
 508             transition ();       /* to PEER1_WRITE_SHUTDOWN */
 509           }
 510         else            /* This will happen during PEER1_HALFCLOSE_READ */
 511           {
 512             peer2.test_ok = GNUNET_YES;
 513             transition ();      /* to PEER1_HALFCLOSE_WRITE_FAIL */
 514           }
 515       }
 516     break;
 517     case PEER1_HALFCLOSE_WRITE_FAIL:
 518       GNUNET_assert (peer == &peer1);
 519       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 520       GNUNET_assert (0 == size);
 521       peer1.test_ok = GNUNET_YES;
 522       break;
 523     case PEER1_READ_SHUTDOWN:
 524       GNUNET_assert (peer == &peer2);
 525       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 526       GNUNET_assert (0 == size);
 527       peer2.test_ok = GNUNET_YES;
 528       break;
 529     case PEER1_WRITE_SHUTDOWN:
 530     case SUCCESS:
 531       GNUNET_assert (0);        /* We shouldn't reach here */
 532     }
 533 }
 534
 535
 536 /**
 537  * Function executed after stream has been established
 538  *
 539  * @param cls the closure from GNUNET_STREAM_open
 540  * @param socket socket to use to communicate with the other side (read/write)
 541  */
 542 static void
 543 stream_open_cb (void *cls,
 544                 struct GNUNET_STREAM_Socket *socket)
 545 {
 546   struct PeerData *peer;
 547
 548   GNUNET_assert (socket == peer1.socket);
 549   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 550               "%s: Stream established from peer1\n",
 551               GNUNET_i2s (&peer1.our_id));
 552   peer = (struct PeerData *) cls;
 553   peer->bytes_wrote = 0;
 554   GNUNET_assert (socket == peer1.socket);
 555   GNUNET_assert (socket == peer->socket);
 556   peer1.test_ok = GNUNET_NO;
 557   peer2.test_ok = GNUNET_NO;
 558   current_test = PEER1_WRITE;
 559   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 560 }
 561
 562
 563 /**
 564  * Input processor
 565  *
 566  * @param cls the closure from GNUNET_STREAM_write/read
 567  * @param status the status of the stream at the time this function is called
 568  * @param data traffic from the other side
 569  * @param size the number of bytes available in data read
 570  * @return number of bytes of processed from 'data' (any data remaining should be
 571  *         given to the next time the read processor is called).
 572  */
 573 static size_t
 574 input_processor (void *cls,
 575                  enum GNUNET_STREAM_Status status,
 576                  const void *input_data,
 577                  size_t size)
 578 {
 579   struct PeerData *peer;
 580
 581   peer = (struct PeerData *) cls;
 582
 583   switch (current_test)
 584     {
 585     case PEER1_WRITE:
 586     case PEER1_HALFCLOSE_READ:
 587       if (GNUNET_STREAM_TIMEOUT == status)
 588         {
 589           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 590                       "Read operation timedout - reading again!\n");
 591           GNUNET_assert (0 == size);
 592           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 593           return 0;
 594         }
 595
 596       GNUNET_assert (GNUNET_STREAM_OK == status);
 597       GNUNET_assert (size <= strlen (data));
 598       GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
 599                                    (const char *) input_data,
 600                                    size));
 601       peer->bytes_read += size;
 602
 603       if (peer->bytes_read < strlen (data))
 604         {
 605           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 606         }
 607       else
 608         {
 609           if (&peer2 == peer) /* Peer2 has completed reading; should write */
 610             {
 611               peer2.test_ok = GNUNET_YES;
 612               transition ();    /* Transition to PEER1_WRITE_SHUTDOWN */
 613             }
 614           else         /* Peer1 has completed reading. End of tests */
 615             {
 616               peer1.test_ok = GNUNET_YES;
 617               transition ();    /* to PEER1_HALFCLOSE_WRITE_FAIL */
 618             }
 619         }
 620       break;
 621     case PEER1_WRITE_SHUTDOWN:
 622       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 623       peer2.test_ok = GNUNET_YES;
 624       break;
 625     case PEER1_HALFCLOSE_WRITE_FAIL:
 626     case PEER1_READ_SHUTDOWN:
 627     case SUCCESS:
 628       GNUNET_assert (0);        /* We shouldn't reach here */
 629     }
 630
 631   return size;
 632 }
 633
 634
 635 /**
 636  * Scheduler call back; to be executed when a new stream is connected
 637  * Called from listen connect for peer2
 638  */
 639 static void
 640 stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 641 {
 642   read_task = GNUNET_SCHEDULER_NO_TASK;
 643   GNUNET_assert (NULL != cls);
 644   peer2.bytes_read = 0;
 645   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 646 }
 647
 648
 649 /**
 650  * Functions of this type are called upon new stream connection from other peers
 651  *
 652  * @param cls the closure from GNUNET_STREAM_listen
 653  * @param socket the socket representing the stream
 654  * @param initiator the identity of the peer who wants to establish a stream
 655  *            with us
 656  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 657  *             stream (the socket will be invalid after the call)
 658  */
 659 static int
 660 stream_listen_cb (void *cls,
 661                   struct GNUNET_STREAM_Socket *socket,
 662                   const struct GNUNET_PeerIdentity *initiator)
 663 {
 664   GNUNET_assert (NULL != socket);
 665   GNUNET_assert (NULL != initiator);
 666   GNUNET_assert (socket != peer1.socket);
 667
 668   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 669               "%s: Peer connected: %s\n",
 670               GNUNET_i2s (&peer2.our_id),
 671               GNUNET_i2s(initiator));
 672
 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, 40), &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   GNUNET_TESTBED_test_run ("test_stream_2peers_halfclose",
 889                            "test_stream_local.conf", NUM_PEERS, event_mask,
 890                            &controller_event_cb, NULL, &test_master, NULL);
 891   if (GNUNET_SYSERR == result)
 892     return 1;
 893   return 0;
 894 }