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   GNUNET_TESTBED_operation_done (peer2.op);
 338 }
 339
 340
 341 /**
 342  * Completion callback for shutdown
 343  *
 344  * @param cls the closure from GNUNET_STREAM_shutdown call
 345  * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
 346  *          SHUT_RDWR)
 347  */
 348 void
 349 shutdown_completion (void *cls,
 350                      int operation)
 351 {
 352   switch (current_test)
 353     {
 354     case PEER1_WRITE:
 355       GNUNET_assert (0);
 356     case PEER1_WRITE_SHUTDOWN:
 357       GNUNET_assert (cls == &peer1);
 358       GNUNET_assert (SHUT_WR == operation);
 359       peer1.test_ok = GNUNET_YES;
 360       /* Peer2 should read with error */
 361       peer2.bytes_read = 0;
 362       GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 363       break;
 364     case PEER1_READ_SHUTDOWN:
 365       peer1.test_ok = GNUNET_YES;
 366       peer2.bytes_wrote = 0;
 367       GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 368       break;
 369     case PEER1_HALFCLOSE_READ:
 370     case PEER1_HALFCLOSE_WRITE_FAIL:
 371     case SUCCESS:
 372       GNUNET_assert (0);        /* We shouldn't reach here */
 373     }
 374 }
 375
 376
 377 /**
 378  * Task for calling STREAM_shutdown
 379  *
 380  * @param cls the peer entity
 381  * @param tc the TaskContext
 382  */
 383 static void
 384 stream_shutdown_task (void *cls,
 385                       const struct GNUNET_SCHEDULER_TaskContext *tc)
 386 {
 387   struct PeerData *peer = cls;
 388
 389   peer->shutdown_handle = GNUNET_STREAM_shutdown (peer->socket,
 390                                                   peer->shutdown_operation,
 391                                                   &shutdown_completion,
 392                                                   peer);
 393   GNUNET_assert (NULL != peer->shutdown_handle);
 394 }
 395
 396
 397 /**
 398  * Something went wrong and timed out. Kill everything and set error flag
 399  */
 400 static void
 401 do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 402 {
 403   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
 404   if (0 != read_task)
 405     {
 406       GNUNET_SCHEDULER_cancel (read_task);
 407     }
 408   result = GNUNET_SYSERR;
 409   abort_task = 0;
 410   do_close (cls, tc);
 411 }
 412
 413
 414 /**
 415  * The transition function; responsible for the transitions among tests
 416  */
 417 static void
 418 transition()
 419 {
 420   if ((GNUNET_YES == peer1.test_ok) && (GNUNET_YES == peer2.test_ok))
 421     {
 422       peer1.test_ok = GNUNET_NO;
 423       peer2.test_ok = GNUNET_NO;
 424       GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 425                   "TEST %d SUCCESSFULL\n", current_test);
 426       switch (current_test)
 427         {
 428         case PEER1_WRITE:
 429           current_test = PEER1_WRITE_SHUTDOWN;
 430           /* Peer1 should shutdown writing */
 431           peer1.shutdown_operation = SHUT_WR;
 432           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 433           break;
 434         case PEER1_WRITE_SHUTDOWN:
 435           current_test = PEER1_HALFCLOSE_READ;
 436           /* Peer2 should be able to write successfully */
 437           peer2.bytes_wrote = 0;
 438           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
 439
 440           /* Peer1 should be able to read successfully */
 441           peer1.bytes_read = 0;
 442           GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
 443           break;
 444         case PEER1_HALFCLOSE_READ:
 445           current_test = PEER1_HALFCLOSE_WRITE_FAIL;
 446           peer1.bytes_wrote = 0;
 447           peer2.bytes_read = 0;
 448           peer2.test_ok = GNUNET_YES;
 449           GNUNET_SCHEDULER_add_now (&stream_write_task, &peer1);
 450           break;
 451         case PEER1_HALFCLOSE_WRITE_FAIL:
 452           current_test = PEER1_READ_SHUTDOWN;
 453           peer1.shutdown_operation = SHUT_RD;
 454           GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
 455           break;
 456         case PEER1_READ_SHUTDOWN:
 457           current_test = SUCCESS;
 458           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 459                       "All tests successful\n");
 460           GNUNET_SCHEDULER_add_now (&do_close, NULL);
 461           break;
 462         case SUCCESS:
 463           GNUNET_assert (0);    /* We shouldn't reach here */
 464
 465         }
 466     }
 467 }
 468
 469 /**
 470  * The write completion function; called upon writing some data to stream or
 471  * upon error
 472  *
 473  * @param cls the closure from GNUNET_STREAM_write/read
 474  * @param status the status of the stream at the time this function is called
 475  * @param size the number of bytes read or written
 476  */
 477 static void
 478 write_completion (void *cls,
 479                   enum GNUNET_STREAM_Status status,
 480                   size_t size)
 481 {
 482   struct PeerData *peer = cls;
 483
 484   switch (current_test)
 485     {
 486     case PEER1_WRITE:
 487     case PEER1_HALFCLOSE_READ:
 488
 489     GNUNET_assert (GNUNET_STREAM_OK == status);
 490     GNUNET_assert (size <= strlen (data));
 491     peer->bytes_wrote += size;
 492
 493     if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
 494       {
 495         GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 496       }
 497     else
 498       {
 499         GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 500                     "Writing completed\n");
 501
 502         if (&peer1 == peer)
 503           {
 504             peer1.test_ok = GNUNET_YES;
 505             transition ();       /* to PEER1_WRITE_SHUTDOWN */
 506           }
 507         else            /* This will happen during PEER1_HALFCLOSE_READ */
 508           {
 509             peer2.test_ok = GNUNET_YES;
 510             transition ();      /* to PEER1_HALFCLOSE_WRITE_FAIL */
 511           }
 512       }
 513     break;
 514     case PEER1_HALFCLOSE_WRITE_FAIL:
 515       GNUNET_assert (peer == &peer1);
 516       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 517       GNUNET_assert (0 == size);
 518       peer1.test_ok = GNUNET_YES;
 519       break;
 520     case PEER1_READ_SHUTDOWN:
 521       GNUNET_assert (peer == &peer2);
 522       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 523       GNUNET_assert (0 == size);
 524       peer2.test_ok = GNUNET_YES;
 525       break;
 526     case PEER1_WRITE_SHUTDOWN:
 527     case SUCCESS:
 528       GNUNET_assert (0);        /* We shouldn't reach here */
 529     }
 530 }
 531
 532
 533 /**
 534  * Function executed after stream has been established
 535  *
 536  * @param cls the closure from GNUNET_STREAM_open
 537  * @param socket socket to use to communicate with the other side (read/write)
 538  */
 539 static void
 540 stream_open_cb (void *cls,
 541                 struct GNUNET_STREAM_Socket *socket)
 542 {
 543   struct PeerData *peer;
 544
 545   GNUNET_assert (socket == peer1.socket);
 546   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 547               "%s: Stream established from peer1\n",
 548               GNUNET_i2s (&peer1.our_id));
 549   peer = (struct PeerData *) cls;
 550   peer->bytes_wrote = 0;
 551   GNUNET_assert (socket == peer1.socket);
 552   GNUNET_assert (socket == peer->socket);
 553   peer1.test_ok = GNUNET_NO;
 554   peer2.test_ok = GNUNET_NO;
 555   current_test = PEER1_WRITE;
 556   GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 557 }
 558
 559
 560 /**
 561  * Input processor
 562  *
 563  * @param cls the closure from GNUNET_STREAM_write/read
 564  * @param status the status of the stream at the time this function is called
 565  * @param data traffic from the other side
 566  * @param size the number of bytes available in data read
 567  * @return number of bytes of processed from 'data' (any data remaining should be
 568  *         given to the next time the read processor is called).
 569  */
 570 static size_t
 571 input_processor (void *cls,
 572                  enum GNUNET_STREAM_Status status,
 573                  const void *input_data,
 574                  size_t size)
 575 {
 576   struct PeerData *peer;
 577
 578   peer = (struct PeerData *) cls;
 579
 580   switch (current_test)
 581     {
 582     case PEER1_WRITE:
 583     case PEER1_HALFCLOSE_READ:
 584       if (GNUNET_STREAM_TIMEOUT == status)
 585         {
 586           GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 587                       "Read operation timedout - reading again!\n");
 588           GNUNET_assert (0 == size);
 589           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 590           return 0;
 591         }
 592
 593       GNUNET_assert (GNUNET_STREAM_OK == status);
 594       GNUNET_assert (size <= strlen (data));
 595       GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
 596                                    (const char *) input_data,
 597                                    size));
 598       peer->bytes_read += size;
 599
 600       if (peer->bytes_read < strlen (data))
 601         {
 602           GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
 603         }
 604       else
 605         {
 606           if (&peer2 == peer) /* Peer2 has completed reading; should write */
 607             {
 608               peer2.test_ok = GNUNET_YES;
 609               transition ();    /* Transition to PEER1_WRITE_SHUTDOWN */
 610             }
 611           else         /* Peer1 has completed reading. End of tests */
 612             {
 613               peer1.test_ok = GNUNET_YES;
 614               transition ();    /* to PEER1_HALFCLOSE_WRITE_FAIL */
 615             }
 616         }
 617       break;
 618     case PEER1_WRITE_SHUTDOWN:
 619       GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
 620       peer2.test_ok = GNUNET_YES;
 621       break;
 622     case PEER1_HALFCLOSE_WRITE_FAIL:
 623     case PEER1_READ_SHUTDOWN:
 624     case SUCCESS:
 625       GNUNET_assert (0);        /* We shouldn't reach here */
 626     }
 627
 628   return size;
 629 }
 630
 631
 632 /**
 633  * Scheduler call back; to be executed when a new stream is connected
 634  * Called from listen connect for peer2
 635  */
 636 static void
 637 stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 638 {
 639   read_task = GNUNET_SCHEDULER_NO_TASK;
 640   GNUNET_assert (NULL != cls);
 641   peer2.bytes_read = 0;
 642   GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
 643 }
 644
 645
 646 /**
 647  * Functions of this type are called upon new stream connection from other peers
 648  *
 649  * @param cls the closure from GNUNET_STREAM_listen
 650  * @param socket the socket representing the stream
 651  * @param initiator the identity of the peer who wants to establish a stream
 652  *            with us
 653  * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
 654  *             stream (the socket will be invalid after the call)
 655  */
 656 static int
 657 stream_listen_cb (void *cls,
 658                   struct GNUNET_STREAM_Socket *socket,
 659                   const struct GNUNET_PeerIdentity *initiator)
 660 {
 661   GNUNET_assert (NULL != socket);
 662   GNUNET_assert (NULL != initiator);
 663   GNUNET_assert (socket != peer1.socket);
 664
 665   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
 666               "%s: Peer connected: %s\n",
 667               GNUNET_i2s (&peer2.our_id),
 668               GNUNET_i2s(initiator));
 669
 670   peer2.socket = socket;
 671   /* FIXME: reading should be done right now instead of a scheduled call */
 672   read_task = GNUNET_SCHEDULER_add_now (&stream_read, (void *) socket);
 673   return GNUNET_OK;
 674 }
 675
 676
 677 /**
 678  * Listen success callback; connects a peer to stream as client
 679  */
 680 static void
 681 stream_connect (void);
 682
 683
 684 /**
 685  * Adapter function called to destroy a connection to
 686  * a service.
 687  *
 688  * @param cls closure
 689  * @param op_result service handle returned from the connect adapter
 690  */
 691 static void
 692 stream_da (void *cls, void *op_result)
 693 {
 694   struct GNUNET_STREAM_ListenSocket *lsocket;
 695
 696   if (&peer2 == cls)
 697   {
 698     lsocket = op_result;
 699     GNUNET_STREAM_listen_close (lsocket);
 700     GNUNET_TESTBED_operation_done (peer1.op);
 701     return;
 702   }
 703   if (&peer1 == cls)
 704   {
 705     GNUNET_assert (op_result == peer1.socket);
 706     GNUNET_STREAM_close (peer1.socket);
 707     GNUNET_SCHEDULER_shutdown (); /* Exit point of the test */
 708     return;
 709   }
 710   GNUNET_assert (0);
 711 }
 712
 713
 714 /**
 715  * Adapter function called to establish a connection to
 716  * a service.
 717  *
 718  * @param cls closure
 719  * @param cfg configuration of the peer to connect to; will be available until
 720  *          GNUNET_TESTBED_operation_done() is called on the operation returned
 721  *          from GNUNET_TESTBED_service_connect()
 722  * @return service handle to return in 'op_result', NULL on error
 723  */
 724 static void *
 725 stream_ca (void *cls, const struct GNUNET_CONFIGURATION_Handle *cfg)
 726 {
 727   struct GNUNET_STREAM_ListenSocket *lsocket;
 728
 729   switch (setup_state)
 730   {
 731   case PEER2_STREAM_CONNECT:
 732     lsocket = GNUNET_STREAM_listen (cfg, 10, &stream_listen_cb, NULL,
 733                                     GNUNET_STREAM_OPTION_SIGNAL_LISTEN_SUCCESS,
 734                                     &stream_connect, GNUNET_STREAM_OPTION_END);
 735     GNUNET_assert (NULL != lsocket);
 736     return lsocket;
 737   case PEER1_STREAM_CONNECT:
 738     peer1.socket = GNUNET_STREAM_open (cfg, &peer2.our_id, 10, &stream_open_cb,
 739                                        &peer1, GNUNET_STREAM_OPTION_END);
 740     GNUNET_assert (NULL != peer1.socket);
 741     return peer1.socket;
 742   default:
 743     GNUNET_assert (0);
 744   }
 745 }
 746
 747
 748 /**
 749  * Listen success callback; connects a peer to stream as client
 750  */
 751 static void
 752 stream_connect (void)
 753 {
 754   GNUNET_assert (PEER2_STREAM_CONNECT == setup_state);
 755   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream listen open successful\n");
 756   peer1.op = GNUNET_TESTBED_service_connect (&peer1, peer1.peer, "stream",
 757                                              NULL, NULL,
 758                                              stream_ca, stream_da, &peer1);
 759   setup_state = PEER1_STREAM_CONNECT;
 760 }
 761
 762
 763 /**
 764  * Callback to be called when the requested peer information is available
 765  *
 766  * @param cb_cls the closure from GNUNET_TETSBED_peer_get_information()
 767  * @param op the operation this callback corresponds to
 768  * @param pinfo the result; will be NULL if the operation has failed
 769  * @param emsg error message if the operation has failed; will be NULL if the
 770  *          operation is successfull
 771  */
 772 static void
 773 peerinfo_cb (void *cb_cls, struct GNUNET_TESTBED_Operation *op_,
 774              const struct GNUNET_TESTBED_PeerInformation *pinfo,
 775              const char *emsg)
 776 {
 777   GNUNET_assert (NULL == emsg);
 778   GNUNET_assert (op == op_);
 779   switch (setup_state)
 780   {
 781   case PEER1_GET_IDENTITY:
 782     memcpy (&peer1.our_id, pinfo->result.id,
 783             sizeof (struct GNUNET_PeerIdentity));
 784     GNUNET_TESTBED_operation_done (op);
 785     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 1 id: %s\n", GNUNET_i2s
 786                 (&peer1.our_id));
 787     op = GNUNET_TESTBED_peer_get_information (peer2.peer,
 788                                               GNUNET_TESTBED_PIT_IDENTITY,
 789                                               &peerinfo_cb, NULL);
 790     setup_state = PEER2_GET_IDENTITY;
 791     break;
 792   case PEER2_GET_IDENTITY:
 793     memcpy (&peer2.our_id, pinfo->result.id,
 794             sizeof (struct GNUNET_PeerIdentity));
 795     GNUNET_TESTBED_operation_done (op);
 796     GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Peer 2 id: %s\n", GNUNET_i2s
 797                 (&peer2.our_id));
 798     peer2.op = GNUNET_TESTBED_service_connect (&peer2, peer2.peer, "stream",
 799                                                NULL, NULL,
 800                                                stream_ca, stream_da, &peer2);
 801     setup_state = PEER2_STREAM_CONNECT;
 802     break;
 803   default:
 804     GNUNET_assert (0);
 805   }
 806 }
 807
 808
 809 /**
 810  * Controller event callback
 811  *
 812  * @param cls NULL
 813  * @param event the controller event
 814  */
 815 static void
 816 controller_event_cb (void *cls,
 817                      const struct GNUNET_TESTBED_EventInformation *event)
 818 {
 819   switch (event->type)
 820   {
 821   case GNUNET_TESTBED_ET_CONNECT:
 822     GNUNET_assert (INIT == setup_state);
 823     GNUNET_TESTBED_operation_done (op);
 824     op = GNUNET_TESTBED_peer_get_information (peer1.peer,
 825                                               GNUNET_TESTBED_PIT_IDENTITY,
 826                                               &peerinfo_cb, NULL);
 827     setup_state = PEER1_GET_IDENTITY;
 828     break;
 829   case GNUNET_TESTBED_ET_OPERATION_FINISHED:
 830     switch (setup_state)
 831     {
 832     case PEER1_STREAM_CONNECT:
 833     case PEER2_STREAM_CONNECT:
 834       GNUNET_assert (NULL == event->details.operation_finished.emsg);
 835       break;
 836     default:
 837       GNUNET_assert (0);
 838     }
 839     break;
 840   default:
 841     GNUNET_assert (0);
 842   }
 843 }
 844
 845
 846 /**
 847  * Signature of a main function for a testcase.
 848  *
 849  * @param cls closure
 850  * @param num_peers number of peers in 'peers'
 851  * @param peers handle to peers run in the testbed
 852  */
 853 static void
 854 test_master (void *cls, unsigned int num_peers,
 855              struct GNUNET_TESTBED_Peer **peers)
 856 {
 857   GNUNET_assert (NULL != peers);
 858   GNUNET_assert (NULL != peers[0]);
 859   GNUNET_assert (NULL != peers[1]);
 860   peer1.peer = peers[0];
 861   peer2.peer = peers[1];
 862   op = GNUNET_TESTBED_overlay_connect (NULL, NULL, NULL, peer2.peer, peer1.peer);
 863   setup_state = INIT;
 864   abort_task =
 865     GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
 866                                   (GNUNET_TIME_UNIT_SECONDS, 40), &do_abort,
 867                                   NULL);
 868 }
 869
 870
 871 /**
 872  * Main function
 873  */
 874 int main (int argc, char **argv)
 875 {
 876   uint64_t event_mask;
 877
 878   result = GNUNET_NO;
 879   event_mask = 0;
 880   event_mask |= (1LL << GNUNET_TESTBED_ET_CONNECT);
 881   event_mask |= (1LL << GNUNET_TESTBED_ET_OPERATION_FINISHED);
 882   GNUNET_TESTBED_test_run ("test_stream_2peers_halfclose",
 883                            "test_stream_local.conf", NUM_PEERS, event_mask,
 884                            &controller_event_cb, NULL, &test_master, NULL);
 885   if (GNUNET_SYSERR == result)
 886     return 1;
 887   return 0;
 888 }