Imported Upstream version 0.9.3upstream/0.9.3

9 files changed, 3258 insertions, 675 deletions

diff --git a/src/stream/Makefile.am b/src/stream/Makefile.am
index 385c0cf..8d74417 100644
--- a/src/stream/Makefile.am
+++ b/src/stream/Makefile.am
@@ -20,8 +20,9 @@ libgnunetstream_la_LDFLAGS = \
   $(GN_LIB_LDFLAGS)
 
 check_PROGRAMS = \
- test_stream_local 
-# test_stream_halfclose
+ test-stream-2peers \
+ test-stream-2peers_halfclose \
+ test-stream-local
 
 EXTRA_DIST = test_stream_local.conf
 
@@ -29,15 +30,23 @@ if ENABLE_TEST_RUN
 TESTS = $(check_PROGRAMS)
 endif
 
+test_stream_2peers_SOURCES = \
+ test_stream_2peers.c
+test_stream_2peers_LDADD = \
+ $(top_builddir)/src/stream/libgnunetstream.la \
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
+
+test_stream_2peers_halfclose_SOURCES = \
+ test_stream_2peers_halfclose.c
+test_stream_2peers_halfclose_LDADD = \
+ $(top_builddir)/src/stream/libgnunetstream.la \
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
+
 test_stream_local_SOURCES = \
  test_stream_local.c
 test_stream_local_LDADD = \
  $(top_builddir)/src/stream/libgnunetstream.la \
- $(top_builddir)/src/util/libgnunetutil.la  
-
-#test_stream_halfclose_SOURCES = \
-# test_stream_halfclose.c
-#test_stream_halfclose_LDADD = \
-# $(top_builddir)/src/stream/libgnunetstream.la \
-# $(top_builddir)/src/util/libgnunetutil.la  
-
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
\ No newline at end of file
diff --git a/src/stream/Makefile.in b/src/stream/Makefile.in
index 44b63f7..9f1c278 100644
--- a/src/stream/Makefile.in
+++ b/src/stream/Makefile.in
@@ -35,7 +35,9 @@ POST_UNINSTALL = :
 build_triplet = @build@
 host_triplet = @host@
 target_triplet = @target@
-check_PROGRAMS = test_stream_local$(EXEEXT)
+check_PROGRAMS = test-stream-2peers$(EXEEXT) \
+	test-stream-2peers_halfclose$(EXEEXT) \
+	test-stream-local$(EXEEXT)
 subdir = src/stream
 DIST_COMMON = README $(srcdir)/Makefile.am $(srcdir)/Makefile.in
 ACLOCAL_M4 = $(top_srcdir)/aclocal.m4
@@ -94,11 +96,26 @@ libgnunetstream_la_LINK = $(LIBTOOL) $(AM_V_lt) --tag=CC \
 	$(AM_LIBTOOLFLAGS) $(LIBTOOLFLAGS) --mode=link $(CCLD) \
 	$(AM_CFLAGS) $(CFLAGS) $(libgnunetstream_la_LDFLAGS) \
 	$(LDFLAGS) -o $@
+am_test_stream_2peers_OBJECTS = test_stream_2peers.$(OBJEXT)
+test_stream_2peers_OBJECTS = $(am_test_stream_2peers_OBJECTS)
+test_stream_2peers_DEPENDENCIES =  \
+	$(top_builddir)/src/stream/libgnunetstream.la \
+	$(top_builddir)/src/util/libgnunetutil.la \
+	$(top_builddir)/src/testing/libgnunettesting.la
+am_test_stream_2peers_halfclose_OBJECTS =  \
+	test_stream_2peers_halfclose.$(OBJEXT)
+test_stream_2peers_halfclose_OBJECTS =  \
+	$(am_test_stream_2peers_halfclose_OBJECTS)
+test_stream_2peers_halfclose_DEPENDENCIES =  \
+	$(top_builddir)/src/stream/libgnunetstream.la \
+	$(top_builddir)/src/util/libgnunetutil.la \
+	$(top_builddir)/src/testing/libgnunettesting.la
 am_test_stream_local_OBJECTS = test_stream_local.$(OBJEXT)
 test_stream_local_OBJECTS = $(am_test_stream_local_OBJECTS)
 test_stream_local_DEPENDENCIES =  \
 	$(top_builddir)/src/stream/libgnunetstream.la \
-	$(top_builddir)/src/util/libgnunetutil.la
+	$(top_builddir)/src/util/libgnunetutil.la \
+	$(top_builddir)/src/testing/libgnunettesting.la
 DEFAULT_INCLUDES = -I.@am__isrc@ -I$(top_builddir)
 depcomp = $(SHELL) $(top_srcdir)/depcomp
 am__depfiles_maybe = depfiles
@@ -125,8 +142,12 @@ am__v_CCLD_0 = @echo "  CCLD  " $@;
 AM_V_GEN = $(am__v_GEN_$(V))
 am__v_GEN_ = $(am__v_GEN_$(AM_DEFAULT_VERBOSITY))
 am__v_GEN_0 = @echo "  GEN   " $@;
-SOURCES = $(libgnunetstream_la_SOURCES) $(test_stream_local_SOURCES)
+SOURCES = $(libgnunetstream_la_SOURCES) $(test_stream_2peers_SOURCES) \
+	$(test_stream_2peers_halfclose_SOURCES) \
+	$(test_stream_local_SOURCES)
 DIST_SOURCES = $(libgnunetstream_la_SOURCES) \
+	$(test_stream_2peers_SOURCES) \
+	$(test_stream_2peers_halfclose_SOURCES) \
 	$(test_stream_local_SOURCES)
 ETAGS = etags
 CTAGS = ctags
@@ -188,6 +209,7 @@ INSTALL_SCRIPT = @INSTALL_SCRIPT@
 INSTALL_STRIP_PROGRAM = @INSTALL_STRIP_PROGRAM@
 INTLLIBS = @INTLLIBS@
 INTL_MACOSX_LIBS = @INTL_MACOSX_LIBS@
+JAVAPORT = @JAVAPORT@
 LD = @LD@
 LDFLAGS = @LDFLAGS@
 LIBADD_DL = @LIBADD_DL@
@@ -221,6 +243,7 @@ LT_DLLOADERS = @LT_DLLOADERS@
 LT_DLPREOPEN = @LT_DLPREOPEN@
 MAKEINFO = @MAKEINFO@
 MKDIR_P = @MKDIR_P@
+MONKEYPREFIX = @MONKEYPREFIX@
 MSGFMT = @MSGFMT@
 MSGFMT_015 = @MSGFMT_015@
 MSGMERGE = @MSGMERGE@
@@ -352,15 +375,31 @@ libgnunetstream_la_LIBADD = \
 libgnunetstream_la_LDFLAGS = \
   $(GN_LIB_LDFLAGS)
 
-# test_stream_halfclose
 EXTRA_DIST = test_stream_local.conf
 @ENABLE_TEST_RUN_TRUE@TESTS = $(check_PROGRAMS)
+test_stream_2peers_SOURCES = \
+ test_stream_2peers.c
+
+test_stream_2peers_LDADD = \
+ $(top_builddir)/src/stream/libgnunetstream.la \
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
+
+test_stream_2peers_halfclose_SOURCES = \
+ test_stream_2peers_halfclose.c
+
+test_stream_2peers_halfclose_LDADD = \
+ $(top_builddir)/src/stream/libgnunetstream.la \
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
+
 test_stream_local_SOURCES = \
  test_stream_local.c
 
 test_stream_local_LDADD = \
  $(top_builddir)/src/stream/libgnunetstream.la \
- $(top_builddir)/src/util/libgnunetutil.la  
+ $(top_builddir)/src/util/libgnunetutil.la \
+ $(top_builddir)/src/testing/libgnunettesting.la
 
 all: all-am
 
@@ -438,8 +477,14 @@ clean-checkPROGRAMS:
 	list=`for p in $$list; do echo "$$p"; done | sed 's/$(EXEEXT)$$//'`; \
 	echo " rm -f" $$list; \
 	rm -f $$list
-test_stream_local$(EXEEXT): $(test_stream_local_OBJECTS) $(test_stream_local_DEPENDENCIES) 
-	@rm -f test_stream_local$(EXEEXT)
+test-stream-2peers$(EXEEXT): $(test_stream_2peers_OBJECTS) $(test_stream_2peers_DEPENDENCIES) 
+	@rm -f test-stream-2peers$(EXEEXT)
+	$(AM_V_CCLD)$(LINK) $(test_stream_2peers_OBJECTS) $(test_stream_2peers_LDADD) $(LIBS)
+test-stream-2peers_halfclose$(EXEEXT): $(test_stream_2peers_halfclose_OBJECTS) $(test_stream_2peers_halfclose_DEPENDENCIES) 
+	@rm -f test-stream-2peers_halfclose$(EXEEXT)
+	$(AM_V_CCLD)$(LINK) $(test_stream_2peers_halfclose_OBJECTS) $(test_stream_2peers_halfclose_LDADD) $(LIBS)
+test-stream-local$(EXEEXT): $(test_stream_local_OBJECTS) $(test_stream_local_DEPENDENCIES) 
+	@rm -f test-stream-local$(EXEEXT)
 	$(AM_V_CCLD)$(LINK) $(test_stream_local_OBJECTS) $(test_stream_local_LDADD) $(LIBS)
 
 mostlyclean-compile:
@@ -449,6 +494,8 @@ distclean-compile:
 	-rm -f *.tab.c
 
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/stream_api.Plo@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/test_stream_2peers.Po@am__quote@
+@AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/test_stream_2peers_halfclose.Po@am__quote@
 @AMDEP_TRUE@@am__include@ @am__quote@./$(DEPDIR)/test_stream_local.Po@am__quote@
 
 .c.o:
@@ -778,12 +825,6 @@ uninstall-am: uninstall-libLTLIBRARIES
 	uninstall-am uninstall-libLTLIBRARIES
 
 
-#test_stream_halfclose_SOURCES = \
-# test_stream_halfclose.c
-#test_stream_halfclose_LDADD = \
-# $(top_builddir)/src/stream/libgnunetstream.la \
-# $(top_builddir)/src/util/libgnunetutil.la  
-
 # Tell versions [3.59,3.63) of GNU make to not export all variables.
 # Otherwise a system limit (for SysV at least) may be exceeded.
 .NOEXPORT:
diff --git a/src/stream/README b/src/stream/README
index 9b550b0..977ca2d 100644
--- a/src/stream/README
+++ b/src/stream/README
@@ -1,11 +1,11 @@
-The aim of the stream library is to provide stream connections between peers in
-GNUnet. This is a convenience library which hides the complexity of dividing
-data stream into packets, transmitting them and retransmitting them in case of
+Stream library provides stream connections between peers in GNUnet. This is a
+convenience library which hides the complexity of dividing data stream into
+packets, transmitting them and retransmitting them in case of communication
 errors.
 
 This library's API are similar to unix PIPE API. The user is expected to open a
 stream to a listening target peer. Once the stream is established, the user can
-use it as a pipe. Any data written into the stream will be readable by the
-target peer.
+use it as a pipe. Any data written into the stream at one peer will be readable
+by the other peer and vice versa.
 
-This library uses mesh API for establishing streams between peers.
\ No newline at end of file
+This library uses mesh API for establishing tunnels between peers.
\ No newline at end of file
diff --git a/src/stream/stream_api.c b/src/stream/stream_api.c
index 84fcdfd..dadba33 100644
--- a/src/stream/stream_api.c
+++ b/src/stream/stream_api.c
@@ -18,17 +18,31 @@
   Boston, MA 02111-1307, USA.
 */
 
+/* TODO:
+ *
+ * Checks for matching the sender and socket->other_peer in server
+ * message handlers  
+ *
+ * Add code for write io timeout
+ *
+ * Include retransmission for control messages
+ **/
+
 /**
  * @file stream/stream_api.c
  * @brief Implementation of the stream library
  * @author Sree Harsha Totakura
  */
+
+
 #include "platform.h"
 #include "gnunet_common.h"
 #include "gnunet_crypto_lib.h"
 #include "gnunet_stream_lib.h"
 #include "stream_protocol.h"
 
+#define LOG(kind,...)                                   \
+  GNUNET_log_from (kind, "stream-api", __VA_ARGS__)
 
 /**
  * The maximum packet size of a stream packet
@@ -36,15 +50,15 @@
 #define MAX_PACKET_SIZE 64000
 
 /**
- * The maximum payload a data message packet can carry
+ * Receive buffer
  */
-static size_t max_payload_size = 
-  MAX_PACKET_SIZE - sizeof (struct GNUNET_STREAM_DataMessage);
+#define RECEIVE_BUFFER_SIZE 4096000
 
 /**
- * Receive buffer
+ * The maximum payload a data message packet can carry
  */
-#define RECEIVE_BUFFER_SIZE 4096000
+static size_t max_payload_size = 
+  MAX_PACKET_SIZE - sizeof (struct GNUNET_STREAM_DataMessage);
 
 /**
  * states in the Protocol
@@ -150,12 +164,6 @@ struct MessageQueue
  */
 struct GNUNET_STREAM_Socket
 {
-
-  /**
-   * The peer identity of the peer at the other end of the stream
-   */
-  struct GNUNET_PeerIdentity other_peer;
-
   /**
    * Retransmission timeout
    */
@@ -177,16 +185,6 @@ struct GNUNET_STREAM_Socket
   struct GNUNET_TIME_Relative ack_time_deadline;
 
   /**
-   * The task for sending timely Acks
-   */
-  GNUNET_SCHEDULER_TaskIdentifier ack_task_id;
-
-  /**
-   * Task scheduled to continue a read operation.
-   */
-  GNUNET_SCHEDULER_TaskIdentifier read_task;
-
-  /**
    * The mesh handle
    */
   struct GNUNET_MESH_Handle *mesh;
@@ -212,6 +210,16 @@ struct GNUNET_STREAM_Socket
   struct GNUNET_MESH_TransmitHandle *transmit_handle;
 
   /**
+   * The current act transmit handle (if a pending ack transmit request exists)
+   */
+  struct GNUNET_MESH_TransmitHandle *ack_transmit_handle;
+
+  /**
+   * Pointer to the current ack message using in ack_task
+   */
+  struct GNUNET_STREAM_AckMessage *ack_msg;
+
+  /**
    * The current message associated with the transmit handle
    */
   struct MessageQueue *queue_head;
@@ -232,14 +240,45 @@ struct GNUNET_STREAM_Socket
   struct GNUNET_STREAM_IOReadHandle *read_handle;
 
   /**
+   * The shutdown handle associated with this socket
+   */
+  struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
+
+  /**
    * Buffer for storing received messages
    */
   void *receive_buffer;
 
   /**
+   * The listen socket from which this socket is derived. Should be NULL if it
+   * is not a derived socket
+   */
+  struct GNUNET_STREAM_ListenSocket *lsocket;
+
+  /**
+   * The peer identity of the peer at the other end of the stream
+   */
+  struct GNUNET_PeerIdentity other_peer;
+
+  /**
    * Task identifier for the read io timeout task
    */
-  GNUNET_SCHEDULER_TaskIdentifier read_io_timeout_task;
+  GNUNET_SCHEDULER_TaskIdentifier read_io_timeout_task_id;
+
+  /**
+   * Task identifier for retransmission task after timeout
+   */
+  GNUNET_SCHEDULER_TaskIdentifier retransmission_timeout_task_id;
+
+  /**
+   * The task for sending timely Acks
+   */
+  GNUNET_SCHEDULER_TaskIdentifier ack_task_id;
+
+  /**
+   * Task scheduled to continue a read operation.
+   */
+  GNUNET_SCHEDULER_TaskIdentifier read_task_id;
 
   /**
    * The state of the protocol associated with this socket
@@ -257,6 +296,11 @@ struct GNUNET_STREAM_Socket
   unsigned int retries;
 
   /**
+   * The application port number (type: uint32_t)
+   */
+  GNUNET_MESH_ApplicationType app_port;
+
+  /**
    * The session id associated with this stream connection
    * FIXME: Not used currently, may be removed
    */
@@ -286,7 +330,7 @@ struct GNUNET_STREAM_Socket
   /**
    * receiver's available buffer after the last acknowledged packet
    */
-  uint32_t receive_window_available;
+  uint32_t receiver_window_available;
 
   /**
    * The offset pointer used during write operation
@@ -310,7 +354,6 @@ struct GNUNET_STREAM_Socket
  */
 struct GNUNET_STREAM_ListenSocket
 {
-
   /**
    * The mesh handle
    */
@@ -328,6 +371,7 @@ struct GNUNET_STREAM_ListenSocket
 
   /**
    * The service port
+   * FIXME: Remove if not required!
    */
   GNUNET_MESH_ApplicationType port;
 };
@@ -339,9 +383,24 @@ struct GNUNET_STREAM_ListenSocket
 struct GNUNET_STREAM_IOWriteHandle
 {
   /**
+   * The socket to which this write handle is associated
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
    * The packet_buffers associated with this Handle
    */
-  struct GNUNET_STREAM_DataMessage *messages[64];
+  struct GNUNET_STREAM_DataMessage *messages[GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH];
+
+  /**
+   * The write continuation callback
+   */
+  GNUNET_STREAM_CompletionContinuation write_cont;
+
+  /**
+   * Write continuation closure
+   */
+  void *write_cont_cls;
 
   /**
    * The bitmap of this IOHandle; Corresponding bit for a message is set when
@@ -350,11 +409,9 @@ struct GNUNET_STREAM_IOWriteHandle
   GNUNET_STREAM_AckBitmap ack_bitmap;
 
   /**
-   * Number of packets sent before waiting for an ack
-   *
-   * FIXME: Do we need this?
+   * Number of bytes in this write handle
    */
-  unsigned int sent_packets;
+  size_t size;
 };
 
 
@@ -376,13 +433,45 @@ struct GNUNET_STREAM_IOReadHandle
 
 
 /**
+ * Handle for Shutdown
+ */
+struct GNUNET_STREAM_ShutdownHandle
+{
+  /**
+   * The socket associated with this shutdown handle
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Shutdown completion callback
+   */
+  GNUNET_STREAM_ShutdownCompletion completion_cb;
+
+  /**
+   * Closure for completion callback
+   */
+  void *completion_cls;
+
+  /**
+   * Close message retransmission task id
+   */
+  GNUNET_SCHEDULER_TaskIdentifier close_msg_retransmission_task_id;
+
+  /**
+   * Which operation to shutdown? SHUT_RD, SHUT_WR or SHUT_RDWR
+   */
+  int operation;  
+};
+
+
+/**
  * Default value in seconds for various timeouts
  */
-static unsigned int default_timeout = 300;
+static unsigned int default_timeout = 10;
 
 
 /**
- * Callback function for sending hello message
+ * Callback function for sending queued message
  *
  * @param cls closure the socket
  * @param size number of bytes available in buf
@@ -401,31 +490,32 @@ send_message_notify (void *cls, size_t size, void *buf)
   if (NULL == head)
     return 0; /* just to be safe */
   if (0 == size)                /* request timed out */
-    {
-      socket->retries++;
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-		  "Message sending timed out. Retry %d \n",
-                  socket->retries);
-      socket->transmit_handle = 
-        GNUNET_MESH_notify_transmit_ready (socket->tunnel,
-                                           0, /* Corking */
-                                           1, /* Priority */
-                                           /* FIXME: exponential backoff */
-                                           socket->retransmit_timeout,
-                                           &socket->other_peer,
-                                           ntohs (head->message->header.size),
-                                           &send_message_notify,
-                                           socket);
-      return 0;
-    }
+  {
+    socket->retries++;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Message sending timed out. Retry %d \n",
+         GNUNET_i2s (&socket->other_peer),
+         socket->retries);
+    socket->transmit_handle = 
+      GNUNET_MESH_notify_transmit_ready (socket->tunnel,
+                                         0, /* Corking */
+                                         1, /* Priority */
+                                         /* FIXME: exponential backoff */
+                                         socket->retransmit_timeout,
+                                         &socket->other_peer,
+                                         ntohs (head->message->header.size),
+                                         &send_message_notify,
+                                         socket);
+    return 0;
+  }
 
   ret = ntohs (head->message->header.size);
   GNUNET_assert (size >= ret);
   memcpy (buf, head->message, ret);
   if (NULL != head->finish_cb)
-    {
-      head->finish_cb (socket, head->finish_cb_cls);
-    }
+  {
+    head->finish_cb (head->finish_cb_cls, socket);
+  }
   GNUNET_CONTAINER_DLL_remove (socket->queue_head,
 			       socket->queue_tail,
 			       head);
@@ -433,19 +523,19 @@ send_message_notify (void *cls, size_t size, void *buf)
   GNUNET_free (head);
   head = socket->queue_head;
   if (NULL != head)    /* more pending messages to send */
-    {
-      socket->retries = 0;
-      socket->transmit_handle = 
-        GNUNET_MESH_notify_transmit_ready (socket->tunnel,
-                                           0, /* Corking */
-                                           1, /* Priority */
-                                           /* FIXME: exponential backoff */
-                                           socket->retransmit_timeout,
-                                           &socket->other_peer,
-                                           ntohs (head->message->header.size),
-                                           &send_message_notify,
-                                           socket);
-    }
+  {
+    socket->retries = 0;
+    socket->transmit_handle = 
+      GNUNET_MESH_notify_transmit_ready (socket->tunnel,
+                                         0, /* Corking */
+                                         1, /* Priority */
+                                         /* FIXME: exponential backoff */
+                                         socket->retransmit_timeout,
+                                         &socket->other_peer,
+                                         ntohs (head->message->header.size),
+                                         &send_message_notify,
+                                         socket);
+  }
   return ret;
 }
 
@@ -466,6 +556,16 @@ queue_message (struct GNUNET_STREAM_Socket *socket,
 {
   struct MessageQueue *queue_entity;
 
+  GNUNET_assert 
+    ((ntohs (message->header.type) >= GNUNET_MESSAGE_TYPE_STREAM_DATA)
+     && (ntohs (message->header.type) <= GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK));
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Queueing message of type %d and size %d\n",
+       GNUNET_i2s (&socket->other_peer),
+       ntohs (message->header.type),
+       ntohs (message->header.size));
+  GNUNET_assert (NULL != message);
   queue_entity = GNUNET_malloc (sizeof (struct MessageQueue));
   queue_entity->message = message;
   queue_entity->finish_cb = finish_cb;
@@ -490,6 +590,31 @@ queue_message (struct GNUNET_STREAM_Socket *socket,
 
 
 /**
+ * Copies a message and queues it for sending using the mesh connection of
+ * given socket 
+ *
+ * @param socket the socket whose mesh connection is used
+ * @param message the message to be sent
+ * @param finish_cb the callback to be called when the message is sent
+ * @param finish_cb_cls the closure for the callback
+ */
+static void
+copy_and_queue_message (struct GNUNET_STREAM_Socket *socket,
+                        const struct GNUNET_STREAM_MessageHeader *message,
+                        SendFinishCallback finish_cb,
+                        void *finish_cb_cls)
+{
+  struct GNUNET_STREAM_MessageHeader *msg_copy;
+  uint16_t size;
+  
+  size = ntohs (message->header.size);
+  msg_copy = GNUNET_malloc (size);
+  memcpy (msg_copy, message, size);
+  queue_message (socket, msg_copy, finish_cb, finish_cb_cls);
+}
+
+
+/**
  * Callback function for sending ack message
  *
  * @param cls closure the ACK message created in ack_task
@@ -500,21 +625,56 @@ queue_message (struct GNUNET_STREAM_Socket *socket,
 static size_t
 send_ack_notify (void *cls, size_t size, void *buf)
 {
-  struct GNUNET_STREAM_AckMessage *ack_msg = cls;
+  struct GNUNET_STREAM_Socket *socket = cls;
 
   if (0 == size)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "%s called with size 0\n", __func__);
-      return 0;
-    }
-  GNUNET_assert (ack_msg->header.header.size <= size);
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s called with size 0\n", __func__);
+    return 0;
+  }
+  GNUNET_assert (ntohs (socket->ack_msg->header.header.size) <= size);
+  
+  size = ntohs (socket->ack_msg->header.header.size);
+  memcpy (buf, socket->ack_msg, size);
   
-  size = ack_msg->header.header.size;
-  memcpy (buf, ack_msg, size);
+  GNUNET_free (socket->ack_msg);
+  socket->ack_msg = NULL;
+  socket->ack_transmit_handle = NULL;
   return size;
 }
 
+/**
+ * Writes data using the given socket. The amount of data written is limited by
+ * the receiver_window_size
+ *
+ * @param socket the socket to use
+ */
+static void 
+write_data (struct GNUNET_STREAM_Socket *socket);
+
+/**
+ * Task for retransmitting data messages if they aren't ACK before their ack
+ * deadline 
+ *
+ * @param cls the socket
+ * @param tc the Task context
+ */
+static void
+retransmission_timeout_task (void *cls,
+                             const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_STREAM_Socket *socket = cls;
+  
+  if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
+    return;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Retransmitting DATA...\n", GNUNET_i2s (&socket->other_peer));
+  socket->retransmission_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
+  write_data (socket);
+}
+
 
 /**
  * Task for sending ACK message
@@ -530,12 +690,10 @@ ack_task (void *cls,
   struct GNUNET_STREAM_AckMessage *ack_msg;
 
   if (GNUNET_SCHEDULER_REASON_SHUTDOWN == tc->reason)
-    {
-      return;
-    }
-
-  socket->ack_task_id = 0;
-
+  {
+    return;
+  }
+  socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
   /* Create the ACK Message */
   ack_msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_AckMessage));
   ack_msg->header.header.size = htons (sizeof (struct 
@@ -545,18 +703,61 @@ ack_task (void *cls,
   ack_msg->base_sequence_number = htonl (socket->read_sequence_number);
   ack_msg->receive_window_remaining = 
     htonl (RECEIVE_BUFFER_SIZE - socket->receive_buffer_size);
-
+  socket->ack_msg = ack_msg;
   /* Request MESH for sending ACK */
-  GNUNET_MESH_notify_transmit_ready (socket->tunnel,
-                                     0, /* Corking */
-                                     1, /* Priority */
-                                     socket->retransmit_timeout,
-                                     &socket->other_peer,
-                                     ntohs (ack_msg->header.header.size),
-                                     &send_ack_notify,
-                                     ack_msg);
+  socket->ack_transmit_handle = 
+    GNUNET_MESH_notify_transmit_ready (socket->tunnel,
+                                       0, /* Corking */
+                                       1, /* Priority */
+                                       socket->retransmit_timeout,
+                                       &socket->other_peer,
+                                       ntohs (ack_msg->header.header.size),
+                                       &send_ack_notify,
+                                       socket);
+}
 
-  
+
+/**
+ * Retransmission task for shutdown messages
+ *
+ * @param cls the shutdown handle
+ * @param tc the Task Context
+ */
+static void
+close_msg_retransmission_task (void *cls,
+                               const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct GNUNET_STREAM_ShutdownHandle *shutdown_handle = cls;
+  struct GNUNET_STREAM_MessageHeader *msg;
+  struct GNUNET_STREAM_Socket *socket;
+
+  GNUNET_assert (NULL != shutdown_handle);
+  socket = shutdown_handle->socket;
+
+  msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
+  msg->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
+  switch (shutdown_handle->operation)
+  {
+  case SHUT_RDWR:
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE);
+    break;
+  case SHUT_RD:
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE);
+    break;
+  case SHUT_WR:
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE);
+    break;
+  default:
+    GNUNET_free (msg);
+    shutdown_handle->close_msg_retransmission_task_id = 
+      GNUNET_SCHEDULER_NO_TASK;
+    return;
+  }
+  queue_message (socket, msg, NULL, NULL);
+  shutdown_handle->close_msg_retransmission_task_id =
+    GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
+                                  &close_msg_retransmission_task,
+                                  shutdown_handle);
 }
 
 
@@ -572,7 +773,7 @@ ackbitmap_modify_bit (GNUNET_STREAM_AckBitmap *bitmap,
 		      unsigned int bit, 
 		      int value)
 {
-  GNUNET_assert (bit < 64);
+  GNUNET_assert (bit < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH);
   if (GNUNET_YES == value)
     *bitmap |= (1LL << bit);
   else
@@ -591,31 +792,14 @@ static uint8_t
 ackbitmap_is_bit_set (const GNUNET_STREAM_AckBitmap *bitmap,
                       unsigned int bit)
 {
-  GNUNET_assert (bit < 64);
+  GNUNET_assert (bit < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH);
   return 0 != (*bitmap & (1LL << bit));
 }
 
 
-
-/**
- * Function called when Data Message is sent
- *
- * @param cls the io_handle corresponding to the Data Message
- * @param socket the socket which was used
- */
-static void
-write_data_finish_cb (void *cls,
-                      struct GNUNET_STREAM_Socket *socket)
-{
-  struct GNUNET_STREAM_IOWriteHandle *io_handle = cls;
-
-  io_handle->sent_packets++;
-}
-
-
 /**
  * Writes data using the given socket. The amount of data written is limited by
- * the receive_window_size
+ * the receiver_window_size
  *
  * @param socket the socket to use
  */
@@ -623,43 +807,59 @@ static void
 write_data (struct GNUNET_STREAM_Socket *socket)
 {
   struct GNUNET_STREAM_IOWriteHandle *io_handle = socket->write_handle;
-  unsigned int packet;
+  int packet;                   /* Although an int, should never be negative */
   int ack_packet;
 
   ack_packet = -1;
   /* Find the last acknowledged packet */
-  for (packet=0; packet < 64; packet++)
-    {      
-      if (GNUNET_YES == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
-                                              packet))
-	ack_packet = packet;        
-      else if (NULL == io_handle->messages[packet])
-	break;
-    }
+  for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+  {      
+    if (GNUNET_YES == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
+                                            packet))
+      ack_packet = packet;        
+    else if (NULL == io_handle->messages[packet])
+      break;
+  }
   /* Resend packets which weren't ack'ed */
   for (packet=0; packet < ack_packet; packet++)
+  {
+    if (GNUNET_NO == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
+                                           packet))
     {
-      if (GNUNET_NO == ackbitmap_is_bit_set (&io_handle->ack_bitmap,
-                                             packet))
-        {
-          queue_message (socket,
-                         &io_handle->messages[packet]->header,
-                         NULL,
-                         NULL);
-        }
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Placing DATA message with sequence %u in send queue\n",
+           GNUNET_i2s (&socket->other_peer),
+           ntohl (io_handle->messages[packet]->sequence_number));
+      copy_and_queue_message (socket,
+                              &io_handle->messages[packet]->header,
+                              NULL,
+                              NULL);
     }
+  }
   packet = ack_packet + 1;
   /* Now send new packets if there is enough buffer space */
   while ( (NULL != io_handle->messages[packet]) &&
-	  (socket->receive_window_available >= ntohs (io_handle->messages[packet]->header.header.size)) )
-    {
-      socket->receive_window_available -= ntohs (io_handle->messages[packet]->header.header.size);
-      queue_message (socket,
-                     &io_handle->messages[packet]->header,
-                     &write_data_finish_cb,
-                     io_handle);
-      packet++;
-    }
+	  (socket->receiver_window_available 
+           >= ntohs (io_handle->messages[packet]->header.header.size)) )
+  {
+    socket->receiver_window_available -= 
+      ntohs (io_handle->messages[packet]->header.header.size);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Placing DATA message with sequence %u in send queue\n",
+         GNUNET_i2s (&socket->other_peer),
+         ntohl (io_handle->messages[packet]->sequence_number));
+    copy_and_queue_message (socket,
+                            &io_handle->messages[packet]->header,
+                            NULL,
+                            NULL);
+    packet++;
+  }
+  if (GNUNET_SCHEDULER_NO_TASK == socket->retransmission_timeout_task_id)
+    socket->retransmission_timeout_task_id = 
+      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply 
+                                    (GNUNET_TIME_UNIT_SECONDS, 8),
+                                    &retransmission_timeout_task,
+                                    socket);
 }
 
 
@@ -671,7 +871,7 @@ write_data (struct GNUNET_STREAM_Socket *socket)
  */
 static void
 call_read_processor (void *cls,
-                          const struct GNUNET_SCHEDULER_TaskContext *tc)
+                     const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_STREAM_Socket *socket = cls;
   size_t read_size;
@@ -680,87 +880,95 @@ call_read_processor (void *cls,
   uint32_t sequence_increase;
   uint32_t offset_increase;
 
-  socket->read_task = GNUNET_SCHEDULER_NO_TASK;
+  socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
   if (0 != (tc->reason & GNUNET_SCHEDULER_REASON_SHUTDOWN))
     return;
 
+  if (NULL == socket->receive_buffer) 
+    return;
+
   GNUNET_assert (NULL != socket->read_handle);
   GNUNET_assert (NULL != socket->read_handle->proc);
 
   /* Check the bitmap for any holes */
   for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
-    {
-      if (GNUNET_NO == ackbitmap_is_bit_set (&socket->ack_bitmap,
-                                             packet))
-        break;
-    }
+  {
+    if (GNUNET_NO == ackbitmap_is_bit_set (&socket->ack_bitmap,
+                                           packet))
+      break;
+  }
   /* We only call read processor if we have the first packet */
   GNUNET_assert (0 < packet);
-
   valid_read_size = 
     socket->receive_buffer_boundaries[packet-1] - socket->copy_offset;
-
   GNUNET_assert (0 != valid_read_size);
-
   /* Cancel the read_io_timeout_task */
-  GNUNET_SCHEDULER_cancel (socket->read_io_timeout_task);
-  socket->read_io_timeout_task = GNUNET_SCHEDULER_NO_TASK;
-
+  GNUNET_SCHEDULER_cancel (socket->read_io_timeout_task_id);
+  socket->read_io_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
   /* Call the data processor */
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Calling read processor\n",
+       GNUNET_i2s (&socket->other_peer));
   read_size = 
     socket->read_handle->proc (socket->read_handle->proc_cls,
                                socket->status,
                                socket->receive_buffer + socket->copy_offset,
                                valid_read_size);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Read processor read %d bytes\n",
+       GNUNET_i2s (&socket->other_peer), read_size);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Read processor completed successfully\n",
+       GNUNET_i2s (&socket->other_peer));
   /* Free the read handle */
   GNUNET_free (socket->read_handle);
   socket->read_handle = NULL;
-
   GNUNET_assert (read_size <= valid_read_size);
   socket->copy_offset += read_size;
-
   /* Determine upto which packet we can remove from the buffer */
   for (packet = 0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+  {
+    if (socket->copy_offset == socket->receive_buffer_boundaries[packet])
+    { packet++; break; }
     if (socket->copy_offset < socket->receive_buffer_boundaries[packet])
       break;
+  }
 
   /* If no packets can be removed we can't move the buffer */
   if (0 == packet) return;
-
   sequence_increase = packet;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Sequence increase after read processor completion: %u\n",
+       GNUNET_i2s (&socket->other_peer), sequence_increase);
 
   /* Shift the data in the receive buffer */
   memmove (socket->receive_buffer,
            socket->receive_buffer 
            + socket->receive_buffer_boundaries[sequence_increase-1],
-           socket->receive_buffer_size - socket->receive_buffer_boundaries[sequence_increase-1]);
-  
+           socket->receive_buffer_size
+           - socket->receive_buffer_boundaries[sequence_increase-1]);
   /* Shift the bitmap */
   socket->ack_bitmap = socket->ack_bitmap >> sequence_increase;
-  
   /* Set read_sequence_number */
   socket->read_sequence_number += sequence_increase;
-  
   /* Set read_offset */
   offset_increase = socket->receive_buffer_boundaries[sequence_increase-1];
   socket->read_offset += offset_increase;
-
   /* Fix copy_offset */
   GNUNET_assert (offset_increase <= socket->copy_offset);
   socket->copy_offset -= offset_increase;
-  
   /* Fix relative boundaries */
   for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+  {
+    if (packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH - sequence_increase)
     {
-      if (packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH - sequence_increase)
-        {
-          socket->receive_buffer_boundaries[packet] = 
-            socket->receive_buffer_boundaries[packet + sequence_increase] 
-            - offset_increase;
-        }
-      else
-        socket->receive_buffer_boundaries[packet] = 0;
+      socket->receive_buffer_boundaries[packet] = 
+        socket->receive_buffer_boundaries[packet + sequence_increase] 
+        - offset_increase;
     }
+    else
+      socket->receive_buffer_boundaries[packet] = 0;
+  }
 }
 
 
@@ -772,20 +980,32 @@ call_read_processor (void *cls,
  */
 static void
 read_io_timeout (void *cls, 
-                const struct GNUNET_SCHEDULER_TaskContext *tc)
+                 const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   struct GNUNET_STREAM_Socket *socket = cls;
+  GNUNET_STREAM_DataProcessor proc;
+  void *proc_cls;
 
-  socket->read_io_timeout_task = GNUNET_SCHEDULER_NO_TASK;
-  if (socket->read_task != GNUNET_SCHEDULER_NO_TASK)
+  socket->read_io_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
+  if (socket->read_task_id != GNUNET_SCHEDULER_NO_TASK)
   {
-    GNUNET_SCHEDULER_cancel (socket->read_task);
-    socket->read_task = GNUNET_SCHEDULER_NO_TASK;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Read task timedout - Cancelling it\n",
+         GNUNET_i2s (&socket->other_peer));
+    GNUNET_SCHEDULER_cancel (socket->read_task_id);
+    socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
   }
   GNUNET_assert (NULL != socket->read_handle);
-  
+  proc = socket->read_handle->proc;
+  proc_cls = socket->read_handle->proc_cls;
+
   GNUNET_free (socket->read_handle);
   socket->read_handle = NULL;
+  /* Call the read processor to signal timeout */
+  proc (proc_cls,
+        GNUNET_STREAM_TIMEOUT,
+        NULL,
+        0);
 }
 
 
@@ -815,96 +1035,150 @@ handle_data (struct GNUNET_STREAM_Socket *socket,
 
   size = htons (msg->header.header.size);
   if (size < sizeof (struct GNUNET_STREAM_DataMessage))
-    {
-      GNUNET_break_op (0);
-      return GNUNET_SYSERR;
-    }
+  {
+    GNUNET_break_op (0);
+    return GNUNET_SYSERR;
+  }
+
+  if (0 != memcmp (sender,
+                   &socket->other_peer,
+                   sizeof (struct GNUNET_PeerIdentity)))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received DATA from non-confirming peer\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_YES;
+  }
 
   switch (socket->state)
+  {
+  case STATE_ESTABLISHED:
+  case STATE_TRANSMIT_CLOSED:
+  case STATE_TRANSMIT_CLOSE_WAIT:
+      
+    /* check if the message's sequence number is in the range we are
+       expecting */
+    relative_sequence_number = 
+      ntohl (msg->sequence_number) - socket->read_sequence_number;
+    if ( relative_sequence_number > GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH)
     {
-    case STATE_ESTABLISHED:
-    case STATE_TRANSMIT_CLOSED:
-    case STATE_TRANSMIT_CLOSE_WAIT:
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Ignoring received message with sequence number %u\n",
+           GNUNET_i2s (&socket->other_peer),
+           ntohl (msg->sequence_number));
+      /* Start ACK sending task if one is not already present */
+      if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
+      {
+        socket->ack_task_id = 
+          GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
+                                        (msg->ack_deadline),
+                                        &ack_task,
+                                        socket);
+      }
+      return GNUNET_YES;
+    }
+      
+    /* Check if we have already seen this message */
+    if (GNUNET_YES == ackbitmap_is_bit_set (&socket->ack_bitmap,
+                                            relative_sequence_number))
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Ignoring already received message with sequence number %u\n",
+           GNUNET_i2s (&socket->other_peer),
+           ntohl (msg->sequence_number));
+      /* Start ACK sending task if one is not already present */
+      if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
+      {
+        socket->ack_task_id = 
+          GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
+                                        (msg->ack_deadline),
+                                        &ack_task,
+                                        socket);
+      }
+      return GNUNET_YES;
+    }
 
-      /* check if the message's sequence number is in the range we are
-         expecting */
-      relative_sequence_number = 
-        ntohl (msg->sequence_number) - socket->read_sequence_number;
-      if ( relative_sequence_number > 64)
-        {
-          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                      "Ignoring received message with sequence number %d",
-                      ntohl (msg->sequence_number));
-          return GNUNET_YES;
-        }
-
-      /* Check if we have to allocate the buffer */
-      size -= sizeof (struct GNUNET_STREAM_DataMessage);
-      relative_offset = ntohl (msg->offset) - socket->read_offset;
-      bytes_needed = relative_offset + size;
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Receiving DATA with sequence number: %u and size: %d from %s\n",
+         GNUNET_i2s (&socket->other_peer),
+         ntohl (msg->sequence_number),
+         ntohs (msg->header.header.size),
+         GNUNET_i2s (&socket->other_peer));
       
-      if (bytes_needed > socket->receive_buffer_size)
-        {
-          if (bytes_needed <= RECEIVE_BUFFER_SIZE)
-            {
-              socket->receive_buffer = GNUNET_realloc (socket->receive_buffer,
-                                                       bytes_needed);
-              socket->receive_buffer_size = bytes_needed;
-            }
-          else
-            {
-              GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                          "Cannot accommodate packet %d as buffer is full\n",
-                          ntohl (msg->sequence_number));
-              return GNUNET_YES;
-            }
-        }
+    /* Check if we have to allocate the buffer */
+    size -= sizeof (struct GNUNET_STREAM_DataMessage);
+    relative_offset = ntohl (msg->offset) - socket->read_offset;
+    bytes_needed = relative_offset + size;
+    if (bytes_needed > socket->receive_buffer_size)
+    {
+      if (bytes_needed <= RECEIVE_BUFFER_SIZE)
+      {
+        socket->receive_buffer = GNUNET_realloc (socket->receive_buffer,
+                                                 bytes_needed);
+        socket->receive_buffer_size = bytes_needed;
+      }
+      else
+      {
+        LOG (GNUNET_ERROR_TYPE_DEBUG,
+             "%s: Cannot accommodate packet %d as buffer is full\n",
+             GNUNET_i2s (&socket->other_peer),
+             ntohl (msg->sequence_number));
+        return GNUNET_YES;
+      }
+    }
       
-      /* Copy Data to buffer */
-      payload = &msg[1];
-      GNUNET_assert (relative_offset + size <= socket->receive_buffer_size);
-      memcpy (socket->receive_buffer + relative_offset,
-              payload,
-              size);
-      socket->receive_buffer_boundaries[relative_sequence_number] = 
-        relative_offset + size;
+    /* Copy Data to buffer */
+    payload = &msg[1];
+    GNUNET_assert (relative_offset + size <= socket->receive_buffer_size);
+    memcpy (socket->receive_buffer + relative_offset,
+            payload,
+            size);
+    socket->receive_buffer_boundaries[relative_sequence_number] = 
+      relative_offset + size;
       
-      /* Modify the ACK bitmap */
-      ackbitmap_modify_bit (&socket->ack_bitmap,
-                            relative_sequence_number,
-                            GNUNET_YES);
+    /* Modify the ACK bitmap */
+    ackbitmap_modify_bit (&socket->ack_bitmap,
+                          relative_sequence_number,
+                          GNUNET_YES);
 
-      /* Start ACK sending task if one is not already present */
-      if (0 == socket->ack_task_id)
-       {
-         socket->ack_task_id = 
-           GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
-                                         (msg->ack_deadline),
-                                         &ack_task,
-                                         socket);
-       }
-
-      if ((NULL != socket->read_handle) /* A read handle is waiting */
-          /* There is no current read task */
-          && (GNUNET_SCHEDULER_NO_TASK == socket->read_task)
-          /* We have the first packet */
-          && (GNUNET_YES == ackbitmap_is_bit_set(&socket->ack_bitmap,
-                                                 0)))
-        {
-          socket->read_task = 
-            GNUNET_SCHEDULER_add_now (&call_read_processor,
+    /* Start ACK sending task if one is not already present */
+    if (GNUNET_SCHEDULER_NO_TASK == socket->ack_task_id)
+    {
+      socket->ack_task_id = 
+        GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_ntoh
+                                      (msg->ack_deadline),
+                                      &ack_task,
                                       socket);
-        }
-      
-      break;
+    }
 
-    default:
-      /* FIXME: call statistics */
-      break;
+    if ((NULL != socket->read_handle) /* A read handle is waiting */
+        /* There is no current read task */
+        && (GNUNET_SCHEDULER_NO_TASK == socket->read_task_id)
+        /* We have the first packet */
+        && (GNUNET_YES == ackbitmap_is_bit_set(&socket->ack_bitmap,
+                                               0)))
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Scheduling read processor\n",
+           GNUNET_i2s (&socket->other_peer));
+          
+      socket->read_task_id = 
+        GNUNET_SCHEDULER_add_now (&call_read_processor,
+                                  socket);
     }
+      
+    break;
+
+  default:
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received data message when it cannot be handled\n",
+         GNUNET_i2s (&socket->other_peer));
+    break;
+  }
   return GNUNET_YES;
 }
 
+
 /**
  * Client's message Handler for GNUNET_MESSAGE_TYPE_STREAM_DATA
  *
@@ -919,11 +1193,11 @@ handle_data (struct GNUNET_STREAM_Socket *socket,
  */
 static int
 client_handle_data (void *cls,
-             struct GNUNET_MESH_Tunnel *tunnel,
-             void **tunnel_ctx,
-             const struct GNUNET_PeerIdentity *sender,
-             const struct GNUNET_MessageHeader *message,
-             const struct GNUNET_ATS_Information*atsi)
+                    struct GNUNET_MESH_Tunnel *tunnel,
+                    void **tunnel_ctx,
+                    const struct GNUNET_PeerIdentity *sender,
+                    const struct GNUNET_MessageHeader *message,
+                    const struct GNUNET_ATS_Information*atsi)
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
@@ -945,10 +1219,31 @@ static void
 set_state_established (void *cls,
                        struct GNUNET_STREAM_Socket *socket)
 {
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Attaining ESTABLISHED state\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, 
+       "%s: Attaining ESTABLISHED state\n",
+       GNUNET_i2s (&socket->other_peer));
   socket->write_offset = 0;
   socket->read_offset = 0;
   socket->state = STATE_ESTABLISHED;
+  /* FIXME: What if listen_cb is NULL */
+  if (NULL != socket->lsocket)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Calling listen callback\n",
+         GNUNET_i2s (&socket->other_peer));
+    if (GNUNET_SYSERR == 
+        socket->lsocket->listen_cb (socket->lsocket->listen_cb_cls,
+                                    socket,
+                                    &socket->other_peer))
+    {
+      socket->state = STATE_CLOSED;
+      /* FIXME: We should close in a decent way */
+      GNUNET_MESH_tunnel_destroy (socket->tunnel); /* Destroy the tunnel */
+      GNUNET_free (socket);
+    }
+  }
+  else if (socket->open_cb)
+    socket->open_cb (socket->open_cls, socket);
 }
 
 
@@ -963,12 +1258,115 @@ set_state_hello_wait (void *cls,
                       struct GNUNET_STREAM_Socket *socket)
 {
   GNUNET_assert (STATE_INIT == socket->state);
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Attaining HELLO_WAIT state\n");
+  LOG (GNUNET_ERROR_TYPE_DEBUG, 
+       "%s: Attaining HELLO_WAIT state\n",
+       GNUNET_i2s (&socket->other_peer));
   socket->state = STATE_HELLO_WAIT;
 }
 
 
 /**
+ * Callback to set state to CLOSE_WAIT
+ *
+ * @param cls the closure from queue_message
+ * @param socket the socket requiring state change
+ */
+static void
+set_state_close_wait (void *cls,
+                      struct GNUNET_STREAM_Socket *socket)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Attaing CLOSE_WAIT state\n",
+       GNUNET_i2s (&socket->other_peer));
+  socket->state = STATE_CLOSE_WAIT;
+  GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
+  socket->receive_buffer = NULL;
+  socket->receive_buffer_size = 0;
+}
+
+
+/**
+ * Callback to set state to RECEIVE_CLOSE_WAIT
+ *
+ * @param cls the closure from queue_message
+ * @param socket the socket requiring state change
+ */
+static void
+set_state_receive_close_wait (void *cls,
+                              struct GNUNET_STREAM_Socket *socket)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Attaing RECEIVE_CLOSE_WAIT state\n",
+       GNUNET_i2s (&socket->other_peer));
+  socket->state = STATE_RECEIVE_CLOSE_WAIT;
+  GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
+  socket->receive_buffer = NULL;
+  socket->receive_buffer_size = 0;
+}
+
+
+/**
+ * Callback to set state to TRANSMIT_CLOSE_WAIT
+ *
+ * @param cls the closure from queue_message
+ * @param socket the socket requiring state change
+ */
+static void
+set_state_transmit_close_wait (void *cls,
+                               struct GNUNET_STREAM_Socket *socket)
+{
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Attaing TRANSMIT_CLOSE_WAIT state\n",
+       GNUNET_i2s (&socket->other_peer));
+  socket->state = STATE_TRANSMIT_CLOSE_WAIT;
+}
+
+
+/**
+ * Callback to set state to CLOSED
+ *
+ * @param cls the closure from queue_message
+ * @param socket the socket requiring state change
+ */
+static void
+set_state_closed (void *cls,
+                  struct GNUNET_STREAM_Socket *socket)
+{
+  socket->state = STATE_CLOSED;
+}
+
+/**
+ * Returns a new HelloAckMessage. Also sets the write sequence number for the
+ * socket
+ *
+ * @param socket the socket for which this HelloAckMessage has to be generated
+ * @return the HelloAckMessage
+ */
+static struct GNUNET_STREAM_HelloAckMessage *
+generate_hello_ack_msg (struct GNUNET_STREAM_Socket *socket)
+{
+  struct GNUNET_STREAM_HelloAckMessage *msg;
+
+  /* Get the random sequence number */
+  socket->write_sequence_number = 
+    GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Generated write sequence number %u\n",
+       GNUNET_i2s (&socket->other_peer),
+       (unsigned int) socket->write_sequence_number);
+  
+  msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
+  msg->header.header.size = 
+    htons (sizeof (struct GNUNET_STREAM_HelloAckMessage));
+  msg->header.header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
+  msg->sequence_number = htonl (socket->write_sequence_number);
+  msg->receiver_window_size = htonl (RECEIVE_BUFFER_SIZE);
+
+  return msg;
+}
+
+
+/**
  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK
  *
  * @param cls the socket (set from GNUNET_MESH_connect)
@@ -992,37 +1390,48 @@ client_handle_hello_ack (void *cls,
   const struct GNUNET_STREAM_HelloAckMessage *ack_msg;
   struct GNUNET_STREAM_HelloAckMessage *reply;
 
+  if (0 != memcmp (sender,
+                   &socket->other_peer,
+                   sizeof (struct GNUNET_PeerIdentity)))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received HELLO_ACK from non-confirming peer\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_YES;
+  }
   ack_msg = (const struct GNUNET_STREAM_HelloAckMessage *) message;
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Received HELLO_ACK from %s\n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+
   GNUNET_assert (socket->tunnel == tunnel);
   switch (socket->state)
   {
   case STATE_HELLO_WAIT:
-      socket->read_sequence_number = ntohl (ack_msg->sequence_number);
-      socket->receive_window_available = ntohl (ack_msg->receive_window_size);
-      /* Get the random sequence number */
-      socket->write_sequence_number = 
-        GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
-      reply = 
-        GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
-      reply->header.header.size = 
-        htons (sizeof (struct GNUNET_STREAM_MessageHeader));
-      reply->header.header.type = 
-        htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
-      reply->sequence_number = htonl (socket->write_sequence_number);
-      reply->receive_window_size = htonl (RECEIVE_BUFFER_SIZE);
-      queue_message (socket, 
-                     &reply->header, 
-                     &set_state_established, 
-                     NULL);      
-      return GNUNET_OK;
+    socket->read_sequence_number = ntohl (ack_msg->sequence_number);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Read sequence number %u\n",
+         GNUNET_i2s (&socket->other_peer),
+         (unsigned int) socket->read_sequence_number);
+    socket->receiver_window_available = ntohl (ack_msg->receiver_window_size);
+    reply = generate_hello_ack_msg (socket);
+    queue_message (socket,
+                   &reply->header, 
+                   &set_state_established, 
+                   NULL);      
+    return GNUNET_OK;
   case STATE_ESTABLISHED:
   case STATE_RECEIVE_CLOSE_WAIT:
     // call statistics (# ACKs ignored++)
     return GNUNET_OK;
   case STATE_INIT:
   default:
-    GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-		"Server sent HELLO_ACK when in state %d\n", socket->state);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Server %s sent HELLO_ACK when in state %d\n", 
+         GNUNET_i2s (&socket->other_peer),
+         GNUNET_i2s (&socket->other_peer),
+         socket->state);
     socket->state = STATE_CLOSED; // introduce STATE_ERROR?
     return GNUNET_SYSERR;
   }
@@ -1050,7 +1459,7 @@ client_handle_reset (void *cls,
                      const struct GNUNET_MessageHeader *message,
                      const struct GNUNET_ATS_Information*atsi)
 {
-  struct GNUNET_STREAM_Socket *socket = cls;
+  // struct GNUNET_STREAM_Socket *socket = cls;
 
   return GNUNET_OK;
 }
@@ -1077,22 +1486,22 @@ handle_transmit_close (struct GNUNET_STREAM_Socket *socket,
   struct GNUNET_STREAM_MessageHeader *reply;
 
   switch (socket->state)
-    {
-    case STATE_ESTABLISHED:
-      socket->state = STATE_RECEIVE_CLOSED;
+  {
+  case STATE_ESTABLISHED:
+    socket->state = STATE_RECEIVE_CLOSED;
 
-      /* Send TRANSMIT_CLOSE_ACK */
-      reply = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
-      reply->header.type = 
-        htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK);
-      reply->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
-      queue_message (socket, reply, NULL, NULL);
-      break;
+    /* Send TRANSMIT_CLOSE_ACK */
+    reply = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
+    reply->header.type = 
+      htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK);
+    reply->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
+    queue_message (socket, reply, NULL, NULL);
+    break;
 
-    default:
-      /* FIXME: Call statistics? */
-      break;
-    }
+  default:
+    /* FIXME: Call statistics? */
+    break;
+  }
   return GNUNET_YES;
 }
 
@@ -1128,6 +1537,141 @@ client_handle_transmit_close (void *cls,
 
 
 /**
+ * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_*_CLOSE_ACK messages
+ *
+ * @param socket the socket
+ * @param tunnel connection to the other end
+ * @param sender who sent the message
+ * @param message the actual message
+ * @param atsi performance data for the connection
+ * @param operation the close operation which is being ACK'ed
+ * @return GNUNET_OK to keep the connection open,
+ *         GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_generic_close_ack (struct GNUNET_STREAM_Socket *socket,
+                          struct GNUNET_MESH_Tunnel *tunnel,
+                          const struct GNUNET_PeerIdentity *sender,
+                          const struct GNUNET_STREAM_MessageHeader *message,
+                          const struct GNUNET_ATS_Information *atsi,
+                          int operation)
+{
+  struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
+
+  shutdown_handle = socket->shutdown_handle;
+  if (NULL == shutdown_handle)
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received CLOSE_ACK when shutdown handle is NULL\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_OK;
+  }
+
+  switch (operation)
+  {
+  case SHUT_RDWR:
+    switch (socket->state)
+    {
+    case STATE_CLOSE_WAIT:
+      if (SHUT_RDWR != shutdown_handle->operation)
+      {
+        LOG (GNUNET_ERROR_TYPE_DEBUG,
+             "%s: Received CLOSE_ACK when shutdown handle is not for "
+             "SHUT_RDWR\n",
+             GNUNET_i2s (&socket->other_peer));
+        return GNUNET_OK;
+      }
+
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received CLOSE_ACK from %s\n",
+           GNUNET_i2s (&socket->other_peer),
+           GNUNET_i2s (&socket->other_peer));
+      socket->state = STATE_CLOSED;
+      break;
+    default:
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received CLOSE_ACK when in it not expected\n",
+           GNUNET_i2s (&socket->other_peer));
+      return GNUNET_OK;
+    }
+    break;
+
+  case SHUT_RD:
+    switch (socket->state)
+    {
+    case STATE_RECEIVE_CLOSE_WAIT:
+      if (SHUT_RD != shutdown_handle->operation)
+      {
+        LOG (GNUNET_ERROR_TYPE_DEBUG,
+             "%s: Received RECEIVE_CLOSE_ACK when shutdown handle "
+             "is not for SHUT_RD\n",
+             GNUNET_i2s (&socket->other_peer));
+        return GNUNET_OK;
+      }
+
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received RECEIVE_CLOSE_ACK from %s\n",
+           GNUNET_i2s (&socket->other_peer),
+           GNUNET_i2s (&socket->other_peer));
+      socket->state = STATE_RECEIVE_CLOSED;
+      break;
+    default:
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received RECEIVE_CLOSE_ACK when in it not expected\n",
+           GNUNET_i2s (&socket->other_peer));
+      return GNUNET_OK;
+    }
+
+    break;
+  case SHUT_WR:
+    switch (socket->state)
+    {
+    case STATE_TRANSMIT_CLOSE_WAIT:
+      if (SHUT_WR != shutdown_handle->operation)
+      {
+        LOG (GNUNET_ERROR_TYPE_DEBUG,
+             "%s: Received TRANSMIT_CLOSE_ACK when shutdown handle "
+             "is not for SHUT_WR\n",
+             GNUNET_i2s (&socket->other_peer));
+        return GNUNET_OK;
+      }
+
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received TRANSMIT_CLOSE_ACK from %s\n",
+           GNUNET_i2s (&socket->other_peer),
+           GNUNET_i2s (&socket->other_peer));
+      socket->state = STATE_TRANSMIT_CLOSED;
+      break;
+    default:
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received TRANSMIT_CLOSE_ACK when in it not expected\n",
+           GNUNET_i2s (&socket->other_peer));
+          
+      return GNUNET_OK;
+    }
+    break;
+  default:
+    GNUNET_assert (0);
+  }
+
+  if (NULL != shutdown_handle->completion_cb) /* Shutdown completion */
+    shutdown_handle->completion_cb(shutdown_handle->completion_cls,
+                                   operation);
+  GNUNET_free (shutdown_handle); /* Free shutdown handle */
+  socket->shutdown_handle = NULL;
+  if (GNUNET_SCHEDULER_NO_TASK
+      != shutdown_handle->close_msg_retransmission_task_id)
+  {
+    GNUNET_SCHEDULER_cancel
+      (shutdown_handle->close_msg_retransmission_task_id);
+    shutdown_handle->close_msg_retransmission_task_id =
+      GNUNET_SCHEDULER_NO_TASK;
+  }
+  return GNUNET_OK;
+}
+
+
+/**
  * Client's message handler for GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE_ACK
  *
  * @param cls the socket (set from GNUNET_MESH_connect)
@@ -1149,6 +1693,67 @@ client_handle_transmit_close_ack (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *)
+                                   message,
+                                   atsi,
+                                   SHUT_WR);
+}
+
+
+/**
+ * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE
+ *
+ * @param socket the socket
+ * @param tunnel connection to the other end
+ * @param sender who sent the message
+ * @param message the actual message
+ * @param atsi performance data for the connection
+ * @return GNUNET_OK to keep the connection open,
+ *         GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_receive_close (struct GNUNET_STREAM_Socket *socket,
+                      struct GNUNET_MESH_Tunnel *tunnel,
+                      const struct GNUNET_PeerIdentity *sender,
+                      const struct GNUNET_STREAM_MessageHeader *message,
+                      const struct GNUNET_ATS_Information *atsi)
+{
+  struct GNUNET_STREAM_MessageHeader *receive_close_ack;
+
+  switch (socket->state)
+  {
+  case STATE_INIT:
+  case STATE_LISTEN:
+  case STATE_HELLO_WAIT:
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Ignoring RECEIVE_CLOSE as it cannot be handled now\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_OK;
+  default:
+    break;
+  }
+  
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Received RECEIVE_CLOSE from %s\n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+  receive_close_ack =
+    GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
+  receive_close_ack->header.size =
+    htons (sizeof (struct GNUNET_STREAM_MessageHeader));
+  receive_close_ack->header.type =
+    htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE_ACK);
+  queue_message (socket,
+                 receive_close_ack,
+                 &set_state_closed,
+                 NULL);
+  
+  /* FIXME: Handle the case where write handle is present; the write operation
+     should be deemed as finised and the write continuation callback
+     has to be called with the stream status GNUNET_STREAM_SHUTDOWN */
   return GNUNET_OK;
 }
 
@@ -1175,7 +1780,12 @@ client_handle_receive_close (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
-  return GNUNET_OK;
+  return
+    handle_receive_close (socket,
+                          tunnel,
+                          sender,
+                          (const struct GNUNET_STREAM_MessageHeader *) message,
+                          atsi);
 }
 
 
@@ -1201,6 +1811,66 @@ client_handle_receive_close_ack (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *)
+                                   message,
+                                   atsi,
+                                   SHUT_RD);
+}
+
+
+/**
+ * Generic handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
+ *
+ * @param socket the socket
+ * @param tunnel connection to the other end
+ * @param sender who sent the message
+ * @param message the actual message
+ * @param atsi performance data for the connection
+ * @return GNUNET_OK to keep the connection open,
+ *         GNUNET_SYSERR to close it (signal serious error)
+ */
+static int
+handle_close (struct GNUNET_STREAM_Socket *socket,
+              struct GNUNET_MESH_Tunnel *tunnel,
+              const struct GNUNET_PeerIdentity *sender,
+              const struct GNUNET_STREAM_MessageHeader *message,
+              const struct GNUNET_ATS_Information*atsi)
+{
+  struct GNUNET_STREAM_MessageHeader *close_ack;
+
+  switch (socket->state)
+  {
+  case STATE_INIT:
+  case STATE_LISTEN:
+  case STATE_HELLO_WAIT:
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Ignoring RECEIVE_CLOSE as it cannot be handled now\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_OK;
+  default:
+    break;
+  }
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Received CLOSE from %s\n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+  close_ack = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
+  close_ack->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
+  close_ack->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE_ACK);
+  queue_message (socket,
+                 close_ack,
+                 &set_state_closed,
+                 NULL);
+  if (socket->state == STATE_CLOSED)
+    return GNUNET_OK;
+
+  GNUNET_free_non_null (socket->receive_buffer); /* Free the receive buffer */
+  socket->receive_buffer = NULL;
+  socket->receive_buffer_size = 0;
   return GNUNET_OK;
 }
 
@@ -1227,7 +1897,11 @@ client_handle_close (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
-  return GNUNET_OK;
+  return handle_close (socket,
+                       tunnel,
+                       sender,
+                       (const struct GNUNET_STREAM_MessageHeader *) message,
+                       atsi);
 }
 
 
@@ -1249,11 +1923,17 @@ client_handle_close_ack (void *cls,
                          void **tunnel_ctx,
                          const struct GNUNET_PeerIdentity *sender,
                          const struct GNUNET_MessageHeader *message,
-                         const struct GNUNET_ATS_Information*atsi)
+                         const struct GNUNET_ATS_Information *atsi)
 {
   struct GNUNET_STREAM_Socket *socket = cls;
 
-  return GNUNET_OK;
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *) 
+                                   message,
+                                   atsi,
+                                   SHUT_RDWR);
 }
 
 /*****************************/
@@ -1313,31 +1993,39 @@ server_handle_hello (void *cls,
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
   struct GNUNET_STREAM_HelloAckMessage *reply;
 
+  if (0 != memcmp (sender,
+                   &socket->other_peer,
+                   sizeof (struct GNUNET_PeerIdentity)))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received HELLO from non-confirming peer\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_YES;
+  }
+
+  GNUNET_assert (GNUNET_MESSAGE_TYPE_STREAM_HELLO == 
+                 ntohs (message->type));
   GNUNET_assert (socket->tunnel == tunnel);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Received HELLO from %s\n", 
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+
   if (STATE_INIT == socket->state)
-    {
-      /* Get the random sequence number */
-      socket->write_sequence_number = 
-        GNUNET_CRYPTO_random_u32 (GNUNET_CRYPTO_QUALITY_NONCE, UINT32_MAX);
-      reply = 
-        GNUNET_malloc (sizeof (struct GNUNET_STREAM_HelloAckMessage));
-      reply->header.header.size = 
-        htons (sizeof (struct GNUNET_STREAM_MessageHeader));
-      reply->header.header.type = 
-        htons (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK);
-      reply->sequence_number = htonl (socket->write_sequence_number);
-      queue_message (socket, 
-		     &reply->header,
-                     &set_state_hello_wait, 
-                     NULL);
-    }
+  {
+    reply = generate_hello_ack_msg (socket);
+    queue_message (socket, 
+                   &reply->header,
+                   &set_state_hello_wait, 
+                   NULL);
+  }
   else
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Client sent HELLO when in state %d\n", socket->state);
-      /* FIXME: Send RESET? */
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Client sent HELLO when in state %d\n", socket->state);
+    /* FIXME: Send RESET? */
       
-    }
+  }
   return GNUNET_OK;
 }
 
@@ -1365,23 +2053,33 @@ server_handle_hello_ack (void *cls,
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
   const struct GNUNET_STREAM_HelloAckMessage *ack_message;
 
-  ack_message = (struct GNUNET_STREAM_HelloAckMessage *) message;
+  GNUNET_assert (GNUNET_MESSAGE_TYPE_STREAM_HELLO_ACK ==
+                 ntohs (message->type));
   GNUNET_assert (socket->tunnel == tunnel);
+  ack_message = (struct GNUNET_STREAM_HelloAckMessage *) message;
   if (STATE_HELLO_WAIT == socket->state)
-    {
-      socket->read_sequence_number = ntohl (ack_message->sequence_number);
-      socket->receive_window_available = 
-        ntohl (ack_message->receive_window_size);
-      /* Attain ESTABLISHED state */
-      set_state_established (NULL, socket);
-    }
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received HELLO_ACK from %s\n",
+         GNUNET_i2s (&socket->other_peer),
+         GNUNET_i2s (&socket->other_peer));
+    socket->read_sequence_number = ntohl (ack_message->sequence_number);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Read sequence number %u\n",
+         GNUNET_i2s (&socket->other_peer),
+         (unsigned int) socket->read_sequence_number);
+    socket->receiver_window_available = 
+      ntohl (ack_message->receiver_window_size);
+    /* Attain ESTABLISHED state */
+    set_state_established (NULL, socket);
+  }
   else
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "Client sent HELLO_ACK when in state %d\n", socket->state);
-      /* FIXME: Send RESET? */
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "Client sent HELLO_ACK when in state %d\n", socket->state);
+    /* FIXME: Send RESET? */
       
-    }
+  }
   return GNUNET_OK;
 }
 
@@ -1406,7 +2104,7 @@ server_handle_reset (void *cls,
                      const struct GNUNET_MessageHeader *message,
                      const struct GNUNET_ATS_Information*atsi)
 {
-  struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
+  // struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
 
   return GNUNET_OK;
 }
@@ -1464,7 +2162,13 @@ server_handle_transmit_close_ack (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
 
-  return GNUNET_OK;
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *)
+                                   message,
+                                   atsi,
+                                   SHUT_WR);
 }
 
 
@@ -1490,7 +2194,12 @@ server_handle_receive_close (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
 
-  return GNUNET_OK;
+  return
+    handle_receive_close (socket,
+                          tunnel,
+                          sender,
+                          (const struct GNUNET_STREAM_MessageHeader *) message,
+                          atsi);
 }
 
 
@@ -1516,14 +2225,21 @@ server_handle_receive_close_ack (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
 
-  return GNUNET_OK;
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *)
+                                   message,
+                                   atsi,
+                                   SHUT_RD);
 }
 
 
 /**
  * Server's message handler for GNUNET_MESSAGE_TYPE_STREAM_CLOSE
  *
- * @param cls the closure
+ * @param cls the listen socket (from GNUNET_MESH_connect in
+ *          GNUNET_STREAM_listen) 
  * @param tunnel connection to the other end
  * @param tunnel_ctx the socket
  * @param sender who sent the message
@@ -1541,8 +2257,12 @@ server_handle_close (void *cls,
                      const struct GNUNET_ATS_Information*atsi)
 {
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
-
-  return GNUNET_OK;
+  
+  return handle_close (socket,
+                       tunnel,
+                       sender,
+                       (const struct GNUNET_STREAM_MessageHeader *) message,
+                       atsi);
 }
 
 
@@ -1568,12 +2288,18 @@ server_handle_close_ack (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = *tunnel_ctx;
 
-  return GNUNET_OK;
+  return handle_generic_close_ack (socket,
+                                   tunnel,
+                                   sender,
+                                   (const struct GNUNET_STREAM_MessageHeader *) 
+                                   message,
+                                   atsi,
+                                   SHUT_RDWR);
 }
 
 
 /**
- * Message Handler for mesh
+ * Handler for DATA_ACK messages
  *
  * @param socket the socket through which the ack was received
  * @param tunnel connection to the other end
@@ -1590,30 +2316,132 @@ handle_ack (struct GNUNET_STREAM_Socket *socket,
 	    const struct GNUNET_STREAM_AckMessage *ack,
 	    const struct GNUNET_ATS_Information*atsi)
 {
+  unsigned int packet;
+  int need_retransmission;
+  
+
+  if (0 != memcmp (sender,
+                   &socket->other_peer,
+                   sizeof (struct GNUNET_PeerIdentity)))
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received ACK from non-confirming peer\n",
+         GNUNET_i2s (&socket->other_peer));
+    return GNUNET_YES;
+  }
+
   switch (socket->state)
+  {
+  case (STATE_ESTABLISHED):
+  case (STATE_RECEIVE_CLOSED):
+  case (STATE_RECEIVE_CLOSE_WAIT):
+    if (NULL == socket->write_handle)
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received DATA_ACK when write_handle is NULL\n",
+           GNUNET_i2s (&socket->other_peer));
+      return GNUNET_OK;
+    }
+    /* FIXME: increment in the base sequence number is breaking current flow
+     */
+    if (!((socket->write_sequence_number 
+           - ntohl (ack->base_sequence_number)) < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH))
+    {
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Received DATA_ACK with unexpected base sequence number\n",
+           GNUNET_i2s (&socket->other_peer));
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Current write sequence: %u; Ack's base sequence: %u\n",
+           GNUNET_i2s (&socket->other_peer),
+           socket->write_sequence_number,
+           ntohl (ack->base_sequence_number));
+      return GNUNET_OK;
+    }
+    /* FIXME: include the case when write_handle is cancelled - ignore the 
+       acks */
+
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: Received DATA_ACK from %s\n",
+         GNUNET_i2s (&socket->other_peer),
+         GNUNET_i2s (&socket->other_peer));
+      
+    /* Cancel the retransmission task */
+    if (GNUNET_SCHEDULER_NO_TASK != socket->retransmission_timeout_task_id)
+    {
+      GNUNET_SCHEDULER_cancel (socket->retransmission_timeout_task_id);
+      socket->retransmission_timeout_task_id = 
+        GNUNET_SCHEDULER_NO_TASK;
+    }
+
+    for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+    {
+      if (NULL == socket->write_handle->messages[packet]) break;
+      if (ntohl (ack->base_sequence_number)
+          >= ntohl (socket->write_handle->messages[packet]->sequence_number))
+        ackbitmap_modify_bit (&socket->write_handle->ack_bitmap,
+                              packet,
+                              GNUNET_YES);
+      else
+        if (GNUNET_YES == 
+            ackbitmap_is_bit_set (&socket->write_handle->ack_bitmap,
+                                  ntohl (socket->write_handle->messages[packet]->sequence_number)
+                                  - ntohl (ack->base_sequence_number)))
+          ackbitmap_modify_bit (&socket->write_handle->ack_bitmap,
+                                packet,
+                                GNUNET_YES);
+    }
+
+    /* Update the receive window remaining
+       FIXME : Should update with the value from a data ack with greater
+       sequence number */
+    socket->receiver_window_available = 
+      ntohl (ack->receive_window_remaining);
+
+    /* Check if we have received all acknowledgements */
+    need_retransmission = GNUNET_NO;
+    for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+    {
+      if (NULL == socket->write_handle->messages[packet]) break;
+      if (GNUNET_YES != ackbitmap_is_bit_set 
+          (&socket->write_handle->ack_bitmap,packet))
+      {
+        need_retransmission = GNUNET_YES;
+        break;
+      }
+    }
+    if (GNUNET_YES == need_retransmission)
     {
-    case (STATE_ESTABLISHED):
-      if (NULL == socket->write_handle)
-        {
-          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                      "Received DATA ACK when write_handle is NULL\n");
-          return GNUNET_OK;
-        }
-
-      socket->write_handle->ack_bitmap = GNUNET_ntohll (ack->bitmap);
-      socket->receive_window_available = 
-        ntohl (ack->receive_window_remaining);
       write_data (socket);
-      break;
-    default:
-      break;
     }
+    else      /* We have to call the write continuation callback now */
+    {
+      /* Free the packets */
+      for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+      {
+        GNUNET_free_non_null (socket->write_handle->messages[packet]);
+      }
+      if (NULL != socket->write_handle->write_cont)
+        socket->write_handle->write_cont
+          (socket->write_handle->write_cont_cls,
+           socket->status,
+           socket->write_handle->size);
+      LOG (GNUNET_ERROR_TYPE_DEBUG,
+           "%s: Write completion callback completed\n",
+           GNUNET_i2s (&socket->other_peer));
+      /* We are done with the write handle - Freeing it */
+      GNUNET_free (socket->write_handle);
+      socket->write_handle = NULL;
+    }
+    break;
+  default:
+    break;
+  }
   return GNUNET_OK;
 }
 
 
 /**
- * Message Handler for mesh
+ * Handler for DATA_ACK messages
  *
  * @param cls the 'struct GNUNET_STREAM_Socket'
  * @param tunnel connection to the other end
@@ -1640,7 +2468,7 @@ client_handle_ack (void *cls,
 
 
 /**
- * Message Handler for mesh
+ * Handler for DATA_ACK messages
  *
  * @param cls the server's listen socket
  * @param tunnel connection to the other end
@@ -1738,19 +2566,21 @@ mesh_peer_connect_callback (void *cls,
 {
   struct GNUNET_STREAM_Socket *socket = cls;
   struct GNUNET_STREAM_MessageHeader *message;
-
-  if (0 != memcmp (&socket->other_peer, 
-                   peer, 
+  
+  if (0 != memcmp (peer,
+                   &socket->other_peer,
                    sizeof (struct GNUNET_PeerIdentity)))
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "A peer (%s) which is not our target has connected to our tunnel", 
-		  GNUNET_i2s (peer));
-      return;
-    }
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: A peer which is not our target has connected to our tunnel\n",
+         GNUNET_i2s(peer));
+    return;
+  }
   
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Target peer %s connected\n", GNUNET_i2s (peer));
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Target peer %s connected\n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
   
   /* Set state to INIT */
   socket->state = STATE_INIT;
@@ -1766,14 +2596,10 @@ mesh_peer_connect_callback (void *cls,
 
   /* Call open callback */
   if (NULL == socket->open_cb)
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-                  "STREAM_open callback is NULL\n");
-    }
-  else
-    {
-      socket->open_cb (socket->open_cls, socket);
-    }
+  {
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "STREAM_open callback is NULL\n");
+  }
 }
 
 
@@ -1787,7 +2613,112 @@ static void
 mesh_peer_disconnect_callback (void *cls,
                                const struct GNUNET_PeerIdentity *peer)
 {
+  struct GNUNET_STREAM_Socket *socket=cls;
+  
+  /* If the state is SHUTDOWN its ok; else set the state of the socket to SYSERR */
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Other peer %s disconnected \n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+}
+
 
+/**
+ * Method called whenever a peer creates a tunnel to us
+ *
+ * @param cls closure
+ * @param tunnel new handle to the tunnel
+ * @param initiator peer that started the tunnel
+ * @param atsi performance information for the tunnel
+ * @return initial tunnel context for the tunnel
+ *         (can be NULL -- that's not an error)
+ */
+static void *
+new_tunnel_notify (void *cls,
+                   struct GNUNET_MESH_Tunnel *tunnel,
+                   const struct GNUNET_PeerIdentity *initiator,
+                   const struct GNUNET_ATS_Information *atsi)
+{
+  struct GNUNET_STREAM_ListenSocket *lsocket = cls;
+  struct GNUNET_STREAM_Socket *socket;
+
+  /* FIXME: If a tunnel is already created, we should not accept new tunnels
+     from the same peer again until the socket is closed */
+
+  socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
+  socket->other_peer = *initiator;
+  socket->tunnel = tunnel;
+  socket->session_id = 0;       /* FIXME */
+  socket->state = STATE_INIT;
+  socket->lsocket = lsocket;
+  
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Peer %s initiated tunnel to us\n", 
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+  
+  /* FIXME: Copy MESH handle from lsocket to socket */
+  
+  return socket;
+}
+
+
+/**
+ * Function called whenever an inbound tunnel is destroyed.  Should clean up
+ * any associated state.  This function is NOT called if the client has
+ * explicitly asked for the tunnel to be destroyed using
+ * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
+ * the tunnel.
+ *
+ * @param cls closure (set from GNUNET_MESH_connect)
+ * @param tunnel connection to the other end (henceforth invalid)
+ * @param tunnel_ctx place where local state associated
+ *                   with the tunnel is stored
+ */
+static void 
+tunnel_cleaner (void *cls,
+                const struct GNUNET_MESH_Tunnel *tunnel,
+                void *tunnel_ctx)
+{
+  struct GNUNET_STREAM_Socket *socket = tunnel_ctx;
+
+  if (tunnel != socket->tunnel)
+    return;
+
+  GNUNET_break_op(0);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: Peer %s has terminated connection abruptly\n",
+       GNUNET_i2s (&socket->other_peer),
+       GNUNET_i2s (&socket->other_peer));
+
+  socket->status = GNUNET_STREAM_SHUTDOWN;
+
+  /* Clear Transmit handles */
+  if (NULL != socket->transmit_handle)
+  {
+    GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
+    socket->transmit_handle = NULL;
+  }
+  if (NULL != socket->ack_transmit_handle)
+  {
+    GNUNET_MESH_notify_transmit_ready_cancel (socket->ack_transmit_handle);
+    GNUNET_free (socket->ack_msg);
+    socket->ack_msg = NULL;
+    socket->ack_transmit_handle = NULL;
+  }
+  /* Stop Tasks using socket->tunnel */
+  if (GNUNET_SCHEDULER_NO_TASK != socket->ack_task_id)
+  {
+    GNUNET_SCHEDULER_cancel (socket->ack_task_id);
+    socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
+  }
+  if (GNUNET_SCHEDULER_NO_TASK != socket->retransmission_timeout_task_id)
+  {
+    GNUNET_SCHEDULER_cancel (socket->retransmission_timeout_task_id);
+    socket->retransmission_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
+  }
+  /* FIXME: Cancel all other tasks using socket->tunnel */
+  socket->tunnel = NULL;
 }
 
 
@@ -1819,55 +2750,167 @@ GNUNET_STREAM_open (const struct GNUNET_CONFIGURATION_Handle *cfg,
 {
   struct GNUNET_STREAM_Socket *socket;
   enum GNUNET_STREAM_Option option;
+  GNUNET_MESH_ApplicationType ports[] = {app_port, 0};
   va_list vargs;                /* Variable arguments */
 
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s\n", __func__);
   socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
   socket->other_peer = *target;
   socket->open_cb = open_cb;
   socket->open_cls = open_cb_cls;
-
   /* Set defaults */
   socket->retransmit_timeout = 
     GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, default_timeout);
-
   va_start (vargs, open_cb_cls); /* Parse variable args */
   do {
     option = va_arg (vargs, enum GNUNET_STREAM_Option);
     switch (option)
-      {
-      case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
-        /* Expect struct GNUNET_TIME_Relative */
-        socket->retransmit_timeout = va_arg (vargs,
-                                             struct GNUNET_TIME_Relative);
-        break;
-      case GNUNET_STREAM_OPTION_END:
-        break;
-      }
+    {
+    case GNUNET_STREAM_OPTION_INITIAL_RETRANSMIT_TIMEOUT:
+      /* Expect struct GNUNET_TIME_Relative */
+      socket->retransmit_timeout = va_arg (vargs,
+                                           struct GNUNET_TIME_Relative);
+      break;
+    case GNUNET_STREAM_OPTION_END:
+      break;
+    }
   } while (GNUNET_STREAM_OPTION_END != option);
   va_end (vargs);               /* End of variable args parsing */
-
   socket->mesh = GNUNET_MESH_connect (cfg, /* the configuration handle */
-                                      1,  /* QUEUE size as parameter? */
+                                      10,  /* QUEUE size as parameter? */
                                       socket, /* cls */
                                       NULL, /* No inbound tunnel handler */
-                                      NULL, /* No inbound tunnel cleaner */
+                                      NULL, /* No in-tunnel cleaner */
                                       client_message_handlers,
-                                      NULL); /* We don't get inbound tunnels */
-  // FIXME: if (NULL == socket->mesh) ...
+                                      ports); /* We don't get inbound tunnels */
+  if (NULL == socket->mesh)   /* Fail if we cannot connect to mesh */
+  {
+    GNUNET_free (socket);
+    return NULL;
+  }
 
   /* Now create the mesh tunnel to target */
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "Creating MESH Tunnel\n");
   socket->tunnel = GNUNET_MESH_tunnel_create (socket->mesh,
                                               NULL, /* Tunnel context */
                                               &mesh_peer_connect_callback,
                                               &mesh_peer_disconnect_callback,
                                               socket);
-  // FIXME: if (NULL == socket->tunnel) ...
-
+  GNUNET_assert (NULL != socket->tunnel);
+  GNUNET_MESH_peer_request_connect_add (socket->tunnel,
+                                        &socket->other_peer);
+  
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s() END\n", __func__);
   return socket;
 }
 
 
 /**
+ * Shutdown the stream for reading or writing (similar to man 2 shutdown).
+ *
+ * @param socket the stream socket
+ * @param operation SHUT_RD, SHUT_WR or SHUT_RDWR
+ * @param completion_cb the callback that will be called upon successful
+ *          shutdown of given operation
+ * @param completion_cls the closure for the completion callback
+ * @return the shutdown handle
+ */
+struct GNUNET_STREAM_ShutdownHandle *
+GNUNET_STREAM_shutdown (struct GNUNET_STREAM_Socket *socket,
+			int operation,
+                        GNUNET_STREAM_ShutdownCompletion completion_cb,
+                        void *completion_cls)
+{
+  struct GNUNET_STREAM_ShutdownHandle *handle;
+  struct GNUNET_STREAM_MessageHeader *msg;
+  
+  GNUNET_assert (NULL == socket->shutdown_handle);
+
+  handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ShutdownHandle));
+  handle->socket = socket;
+  handle->completion_cb = completion_cb;
+  handle->completion_cls = completion_cls;
+  socket->shutdown_handle = handle;
+
+  msg = GNUNET_malloc (sizeof (struct GNUNET_STREAM_MessageHeader));
+  msg->header.size = htons (sizeof (struct GNUNET_STREAM_MessageHeader));
+  switch (operation)
+  {
+  case SHUT_RD:
+    handle->operation = SHUT_RD;
+    if (NULL != socket->read_handle)
+      LOG (GNUNET_ERROR_TYPE_WARNING,
+           "Existing read handle should be cancelled before shutting"
+           " down reading\n");
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_RECEIVE_CLOSE);
+    queue_message (socket,
+                   msg,
+                   &set_state_receive_close_wait,
+                   NULL);
+    break;
+  case SHUT_WR:
+    handle->operation = SHUT_WR;
+    if (NULL != socket->write_handle)
+      LOG (GNUNET_ERROR_TYPE_WARNING,
+           "Existing write handle should be cancelled before shutting"
+           " down writing\n");
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_TRANSMIT_CLOSE);
+    queue_message (socket,
+                   msg,
+                   &set_state_transmit_close_wait,
+                   NULL);
+    break;
+  case SHUT_RDWR:
+    handle->operation = SHUT_RDWR;
+    if (NULL != socket->write_handle)
+      LOG (GNUNET_ERROR_TYPE_WARNING,
+           "Existing write handle should be cancelled before shutting"
+           " down writing\n");
+    if (NULL != socket->read_handle)
+      LOG (GNUNET_ERROR_TYPE_WARNING,
+           "Existing read handle should be cancelled before shutting"
+           " down reading\n");
+    msg->header.type = htons (GNUNET_MESSAGE_TYPE_STREAM_CLOSE);
+    queue_message (socket,
+                   msg,
+                   &set_state_close_wait,
+                   NULL);
+    break;
+  default:
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+         "GNUNET_STREAM_shutdown called with invalid value for "
+         "parameter operation -- Ignoring\n");
+    GNUNET_free (msg);
+    GNUNET_free (handle);
+    return NULL;
+  }
+  handle->close_msg_retransmission_task_id =
+    GNUNET_SCHEDULER_add_delayed (socket->retransmit_timeout,
+                                  &close_msg_retransmission_task,
+                                  handle);
+  return handle;
+}
+
+
+/**
+ * Cancels a pending shutdown
+ *
+ * @param handle the shutdown handle returned from GNUNET_STREAM_shutdown
+ */
+void
+GNUNET_STREAM_shutdown_cancel (struct GNUNET_STREAM_ShutdownHandle *handle)
+{
+  if (GNUNET_SCHEDULER_NO_TASK != handle->close_msg_retransmission_task_id)
+    GNUNET_SCHEDULER_cancel (handle->close_msg_retransmission_task_id);
+  GNUNET_free (handle);
+  return;
+}
+
+
+/**
  * Closes the stream
  *
  * @param socket the stream socket
@@ -1877,20 +2920,45 @@ GNUNET_STREAM_close (struct GNUNET_STREAM_Socket *socket)
 {
   struct MessageQueue *head;
 
-  if (socket->read_task != GNUNET_SCHEDULER_NO_TASK)
+  if (NULL != socket->read_handle)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+	 "Closing STREAM socket when a read handle is pending\n");
+  }
+  if (NULL != socket->write_handle)
+  {
+    LOG (GNUNET_ERROR_TYPE_WARNING,
+	 "Closing STREAM socket when a write handle is pending\n");
+  }
+
+  if (socket->read_task_id != GNUNET_SCHEDULER_NO_TASK)
   {
     /* socket closed with read task pending!? */
     GNUNET_break (0);
-    GNUNET_SCHEDULER_cancel (socket->read_task);
-    socket->read_task = GNUNET_SCHEDULER_NO_TASK;
+    GNUNET_SCHEDULER_cancel (socket->read_task_id);
+    socket->read_task_id = GNUNET_SCHEDULER_NO_TASK;
+  }
+  
+  /* Terminate the ack'ing tasks if they are still present */
+  if (socket->ack_task_id != GNUNET_SCHEDULER_NO_TASK)
+  {
+    GNUNET_SCHEDULER_cancel (socket->ack_task_id);
+    socket->ack_task_id = GNUNET_SCHEDULER_NO_TASK;
   }
 
   /* Clear Transmit handles */
   if (NULL != socket->transmit_handle)
-    {
-      GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
-      socket->transmit_handle = NULL;
-    }
+  {
+    GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
+    socket->transmit_handle = NULL;
+  }
+  if (NULL != socket->ack_transmit_handle)
+  {
+    GNUNET_MESH_notify_transmit_ready_cancel (socket->ack_transmit_handle);
+    GNUNET_free (socket->ack_msg);
+    socket->ack_msg = NULL;
+    socket->ack_transmit_handle = NULL;
+  }
 
   /* Clear existing message queue */
   while (NULL != (head = socket->queue_head)) {
@@ -1903,101 +2971,29 @@ GNUNET_STREAM_close (struct GNUNET_STREAM_Socket *socket)
 
   /* Close associated tunnel */
   if (NULL != socket->tunnel)
-    {
-      GNUNET_MESH_tunnel_destroy (socket->tunnel);
-      socket->tunnel = NULL;
-    }
+  {
+    GNUNET_MESH_tunnel_destroy (socket->tunnel);
+    socket->tunnel = NULL;
+  }
 
   /* Close mesh connection */
-  if (NULL != socket->mesh)
-    {
-      GNUNET_MESH_disconnect (socket->mesh);
-      socket->mesh = NULL;
-    }
+  if (NULL != socket->mesh && NULL == socket->lsocket)
+  {
+    GNUNET_MESH_disconnect (socket->mesh);
+    socket->mesh = NULL;
+  }
   
   /* Release receive buffer */
   if (NULL != socket->receive_buffer)
-    {
-      GNUNET_free (socket->receive_buffer);
-    }
+  {
+    GNUNET_free (socket->receive_buffer);
+  }
 
   GNUNET_free (socket);
 }
 
 
 /**
- * Method called whenever a peer creates a tunnel to us
- *
- * @param cls closure
- * @param tunnel new handle to the tunnel
- * @param initiator peer that started the tunnel
- * @param atsi performance information for the tunnel
- * @return initial tunnel context for the tunnel
- *         (can be NULL -- that's not an error)
- */
-static void *
-new_tunnel_notify (void *cls,
-                   struct GNUNET_MESH_Tunnel *tunnel,
-                   const struct GNUNET_PeerIdentity *initiator,
-                   const struct GNUNET_ATS_Information *atsi)
-{
-  struct GNUNET_STREAM_ListenSocket *lsocket = cls;
-  struct GNUNET_STREAM_Socket *socket;
-
-  socket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_Socket));
-  socket->tunnel = tunnel;
-  socket->session_id = 0;       /* FIXME */
-  socket->other_peer = *initiator;
-  socket->state = STATE_INIT;
-
-  if (GNUNET_SYSERR == lsocket->listen_cb (lsocket->listen_cb_cls,
-                                           socket,
-                                           &socket->other_peer))
-    {
-      socket->state = STATE_CLOSED;
-      /* FIXME: Send CLOSE message and then free */
-      GNUNET_free (socket);
-      GNUNET_MESH_tunnel_destroy (tunnel); /* Destroy the tunnel */
-    }
-  return socket;
-}
-
-
-/**
- * Function called whenever an inbound tunnel is destroyed.  Should clean up
- * any associated state.  This function is NOT called if the client has
- * explicitly asked for the tunnel to be destroyed using
- * GNUNET_MESH_tunnel_destroy. It must NOT call GNUNET_MESH_tunnel_destroy on
- * the tunnel.
- *
- * @param cls closure (set from GNUNET_MESH_connect)
- * @param tunnel connection to the other end (henceforth invalid)
- * @param tunnel_ctx place where local state associated
- *                   with the tunnel is stored
- */
-static void 
-tunnel_cleaner (void *cls,
-                const struct GNUNET_MESH_Tunnel *tunnel,
-                void *tunnel_ctx)
-{
-  struct GNUNET_STREAM_Socket *socket = tunnel_ctx;
-  
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
-              "Peer %s has terminated connection abruptly\n",
-              GNUNET_i2s (&socket->other_peer));
-
-  socket->status = GNUNET_STREAM_SHUTDOWN;
-  /* Clear Transmit handles */
-  if (NULL != socket->transmit_handle)
-    {
-      GNUNET_MESH_notify_transmit_ready_cancel (socket->transmit_handle);
-      socket->transmit_handle = NULL;
-    }
-  socket->tunnel = NULL;
-}
-
-
-/**
  * Listens for stream connections for a specific application ports
  *
  * @param cfg the configuration to use
@@ -2015,10 +3011,8 @@ GNUNET_STREAM_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
 {
   /* FIXME: Add variable args for passing configration options? */
   struct GNUNET_STREAM_ListenSocket *lsocket;
-  GNUNET_MESH_ApplicationType app_types[2];
+  GNUNET_MESH_ApplicationType ports[] = {app_port, 0};
 
-  app_types[0] = app_port;
-  app_types[1] = 0;
   lsocket = GNUNET_malloc (sizeof (struct GNUNET_STREAM_ListenSocket));
   lsocket->port = app_port;
   lsocket->listen_cb = listen_cb;
@@ -2029,7 +3023,8 @@ GNUNET_STREAM_listen (const struct GNUNET_CONFIGURATION_Handle *cfg,
                                        &new_tunnel_notify,
                                        &tunnel_cleaner,
                                        server_message_handlers,
-                                       app_types);
+                                       ports);
+  GNUNET_assert (NULL != lsocket->mesh);
   return lsocket;
 }
 
@@ -2043,6 +3038,7 @@ void
 GNUNET_STREAM_listen_close (struct GNUNET_STREAM_ListenSocket *lsocket)
 {
   /* Close MESH connection */
+  GNUNET_assert (NULL != lsocket->mesh);
   GNUNET_MESH_disconnect (lsocket->mesh);
   
   GNUNET_free (lsocket);
@@ -2050,15 +3046,22 @@ GNUNET_STREAM_listen_close (struct GNUNET_STREAM_ListenSocket *lsocket)
 
 
 /**
- * Tries to write the given data to the stream
+ * Tries to write the given data to the stream. The maximum size of data that
+ * can be written as part of a write operation is (64 * (64000 - sizeof (struct
+ * GNUNET_STREAM_DataMessage))). If size is greater than this it is not an API
+ * violation, however only the said number of maximum bytes will be written.
  *
  * @param socket the socket representing a stream
  * @param data the data buffer from where the data is written into the stream
  * @param size the number of bytes to be written from the data buffer
  * @param timeout the timeout period
- * @param write_cont the function to call upon writing some bytes into the stream
+ * @param write_cont the function to call upon writing some bytes into the
+ *          stream 
  * @param write_cont_cls the closure
- * @return handle to cancel the operation
+ *
+ * @return handle to cancel the operation; if a previous write is pending or
+ *           the stream has been shutdown for this operation then write_cont is
+ *           immediately called and NULL is returned.
  */
 struct GNUNET_STREAM_IOWriteHandle *
 GNUNET_STREAM_write (struct GNUNET_STREAM_Socket *socket,
@@ -2075,6 +3078,10 @@ GNUNET_STREAM_write (struct GNUNET_STREAM_Socket *socket,
   uint32_t payload_size;
   struct GNUNET_STREAM_DataMessage *data_msg;
   const void *sweep;
+  struct GNUNET_TIME_Relative ack_deadline;
+
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s\n", __func__);
 
   /* Return NULL if there is already a write request pending */
   if (NULL != socket->write_handle)
@@ -2082,70 +3089,101 @@ GNUNET_STREAM_write (struct GNUNET_STREAM_Socket *socket,
     GNUNET_break (0);
     return NULL;
   }
-  if (!((STATE_ESTABLISHED == socket->state)
-        || (STATE_RECEIVE_CLOSE_WAIT == socket->state)
-        || (STATE_RECEIVE_CLOSED == socket->state)))
-    {
-      GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
-                  "Attempting to write on a closed (OR) not-yet-established"
-                  "stream\n"); 
-      return NULL;
-    } 
+
+  switch (socket->state)
+  {
+  case STATE_TRANSMIT_CLOSED:
+  case STATE_TRANSMIT_CLOSE_WAIT:
+  case STATE_CLOSED:
+  case STATE_CLOSE_WAIT:
+    if (NULL != write_cont)
+      write_cont (write_cont_cls, GNUNET_STREAM_SHUTDOWN, 0);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s() END\n", __func__);
+    return NULL;
+  case STATE_INIT:
+  case STATE_LISTEN:
+  case STATE_HELLO_WAIT:
+    if (NULL != write_cont)
+      /* FIXME: GNUNET_STREAM_SYSERR?? */
+      write_cont (write_cont_cls, GNUNET_STREAM_SYSERR, 0);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s() END\n", __func__);
+    return NULL;
+  case STATE_ESTABLISHED:
+  case STATE_RECEIVE_CLOSED:
+  case STATE_RECEIVE_CLOSE_WAIT:
+    break;
+  }
+
   if (GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH * max_payload_size < size)
     size = GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH  * max_payload_size;
   num_needed_packets = (size + (max_payload_size - 1)) / max_payload_size;
   io_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOWriteHandle));
+  io_handle->socket = socket;
+  io_handle->write_cont = write_cont;
+  io_handle->write_cont_cls = write_cont_cls;
+  io_handle->size = size;
   sweep = data;
+  /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
+     determined from RTT */
+  ack_deadline = GNUNET_TIME_relative_multiply (GNUNET_TIME_UNIT_SECONDS, 5);
   /* Divide the given buffer into packets for sending */
   for (packet=0; packet < num_needed_packets; packet++)
+  {
+    if ((packet + 1) * max_payload_size < size) 
+    {
+      payload_size = max_payload_size;
+      packet_size = MAX_PACKET_SIZE;
+    }
+    else 
     {
-      if ((packet + 1) * max_payload_size < size) 
-        {
-          payload_size = max_payload_size;
-          packet_size = MAX_PACKET_SIZE;
-        }
-      else 
-        {
-          payload_size = size - packet * max_payload_size;
-          packet_size =  payload_size + sizeof (struct
-                                                GNUNET_STREAM_DataMessage); 
-        }
-      io_handle->messages[packet] = GNUNET_malloc (packet_size);
-      io_handle->messages[packet]->header.header.size = htons (packet_size);
-      io_handle->messages[packet]->header.header.type =
-        htons (GNUNET_MESSAGE_TYPE_STREAM_DATA);
-      io_handle->messages[packet]->sequence_number =
-        htons (socket->write_sequence_number++);
-      io_handle->messages[packet]->offset = htons (socket->write_offset);
-
-      /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
-         determined from RTT */
-      io_handle->messages[packet]->ack_deadline = 
-        GNUNET_TIME_relative_hton (GNUNET_TIME_relative_multiply 
-                                   (GNUNET_TIME_UNIT_SECONDS, 5));
-      data_msg = io_handle->messages[packet];
-      /* Copy data from given buffer to the packet */
-      memcpy (&data_msg[1],
-              sweep,
-              payload_size);
-      sweep += payload_size;
-      socket->write_offset += payload_size;
+      payload_size = size - packet * max_payload_size;
+      packet_size =  payload_size + sizeof (struct
+                                            GNUNET_STREAM_DataMessage); 
     }
+    io_handle->messages[packet] = GNUNET_malloc (packet_size);
+    io_handle->messages[packet]->header.header.size = htons (packet_size);
+    io_handle->messages[packet]->header.header.type =
+      htons (GNUNET_MESSAGE_TYPE_STREAM_DATA);
+    io_handle->messages[packet]->sequence_number =
+      htonl (socket->write_sequence_number++);
+    io_handle->messages[packet]->offset = htonl (socket->write_offset);
+
+    /* FIXME: Remove the fixed delay for ack deadline; Set it to the value
+       determined from RTT */
+    io_handle->messages[packet]->ack_deadline =
+      GNUNET_TIME_relative_hton (ack_deadline);
+    data_msg = io_handle->messages[packet];
+    /* Copy data from given buffer to the packet */
+    memcpy (&data_msg[1],
+            sweep,
+            payload_size);
+    sweep += payload_size;
+    socket->write_offset += payload_size;
+  }
   socket->write_handle = io_handle;
   write_data (socket);
 
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s() END\n", __func__);
+
   return io_handle;
 }
 
 
+
 /**
- * Tries to read data from the stream
+ * Tries to read data from the stream.
  *
  * @param socket the socket representing a stream
  * @param timeout the timeout period
  * @param proc function to call with data (once only)
  * @param proc_cls the closure for proc
- * @return handle to cancel the operation
+ *
+ * @return handle to cancel the operation; if the stream has been shutdown for
+ *           this type of opeartion then the DataProcessor is immediately
+ *           called with GNUNET_STREAM_SHUTDOWN as status and NULL if returned
  */
 struct GNUNET_STREAM_IOReadHandle *
 GNUNET_STREAM_read (struct GNUNET_STREAM_Socket *socket,
@@ -2155,26 +3193,99 @@ GNUNET_STREAM_read (struct GNUNET_STREAM_Socket *socket,
 {
   struct GNUNET_STREAM_IOReadHandle *read_handle;
   
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: %s()\n", 
+       GNUNET_i2s (&socket->other_peer),
+       __func__);
+
   /* Return NULL if there is already a read handle; the user has to cancel that
-  first before continuing or has to wait until it is completed */
+     first before continuing or has to wait until it is completed */
   if (NULL != socket->read_handle) return NULL;
 
+  GNUNET_assert (NULL != proc);
+
+  switch (socket->state)
+  {
+  case STATE_RECEIVE_CLOSED:
+  case STATE_RECEIVE_CLOSE_WAIT:
+  case STATE_CLOSED:
+  case STATE_CLOSE_WAIT:
+    proc (proc_cls, GNUNET_STREAM_SHUTDOWN, NULL, 0);
+    LOG (GNUNET_ERROR_TYPE_DEBUG,
+         "%s: %s() END\n",
+         GNUNET_i2s (&socket->other_peer),
+         __func__);
+    return NULL;
+  default:
+    break;
+  }
+
   read_handle = GNUNET_malloc (sizeof (struct GNUNET_STREAM_IOReadHandle));
   read_handle->proc = proc;
+  read_handle->proc_cls = proc_cls;
   socket->read_handle = read_handle;
 
   /* Check if we have a packet at bitmap 0 */
   if (GNUNET_YES == ackbitmap_is_bit_set (&socket->ack_bitmap,
                                           0))
-    {
-      socket->read_task = GNUNET_SCHEDULER_add_now (&call_read_processor,
-                                                    socket);
+  {
+    socket->read_task_id = GNUNET_SCHEDULER_add_now (&call_read_processor,
+                                                     socket);
    
-    }
+  }
   
   /* Setup the read timeout task */
-  socket->read_io_timeout_task = GNUNET_SCHEDULER_add_delayed (timeout,
-                                                               &read_io_timeout,
-                                                               socket);
+  socket->read_io_timeout_task_id =
+    GNUNET_SCHEDULER_add_delayed (timeout,
+                                  &read_io_timeout,
+                                  socket);
+  LOG (GNUNET_ERROR_TYPE_DEBUG,
+       "%s: %s() END\n",
+       GNUNET_i2s (&socket->other_peer),
+       __func__);
   return read_handle;
 }
+
+
+/**
+ * Cancel pending write operation.
+ *
+ * @param ioh handle to operation to cancel
+ */
+void
+GNUNET_STREAM_io_write_cancel (struct GNUNET_STREAM_IOWriteHandle *ioh)
+{
+  struct GNUNET_STREAM_Socket *socket = ioh->socket;
+  unsigned int packet;
+
+  GNUNET_assert (NULL != socket->write_handle);
+  GNUNET_assert (socket->write_handle == ioh);
+
+  if (GNUNET_SCHEDULER_NO_TASK != socket->retransmission_timeout_task_id)
+  {
+    GNUNET_SCHEDULER_cancel (socket->retransmission_timeout_task_id);
+    socket->retransmission_timeout_task_id = GNUNET_SCHEDULER_NO_TASK;
+  }
+
+  for (packet=0; packet < GNUNET_STREAM_ACK_BITMAP_BIT_LENGTH; packet++)
+  {
+    if (NULL == ioh->messages[packet]) break;
+    GNUNET_free (ioh->messages[packet]);
+  }
+      
+  GNUNET_free (socket->write_handle);
+  socket->write_handle = NULL;
+  return;
+}
+
+
+/**
+ * Cancel pending read operation.
+ *
+ * @param ioh handle to operation to cancel
+ */
+void
+GNUNET_STREAM_io_read_cancel (struct GNUNET_STREAM_IOReadHandle *ioh)
+{
+  return;
+}
diff --git a/src/stream/stream_protocol.h b/src/stream/stream_protocol.h
index 0c1987e..0c5376f 100644
--- a/src/stream/stream_protocol.h
+++ b/src/stream/stream_protocol.h
@@ -127,10 +127,8 @@ struct GNUNET_STREAM_AckMessage
   GNUNET_STREAM_AckBitmap bitmap GNUNET_PACKED;
 
   /**
-   * The sequence number of the Data Message upto which the receiver has filled
-   * its buffer without any missing packets
-   *
-   * FIXME: Do we need this?
+   * The sequence number of the next Data Message receiver is
+   * anticipating. Data messages less than this number are received by receiver
    */
   uint32_t base_sequence_number GNUNET_PACKED;
 
@@ -163,7 +161,7 @@ struct GNUNET_STREAM_HelloAckMessage
    *
    * FIXME: Remove if not needed
    */
-  uint32_t receive_window_size;
+  uint32_t receiver_window_size;
 };
 
 
diff --git a/src/stream/test_stream_2peers.c b/src/stream/test_stream_2peers.c
new file mode 100644
index 0000000..1fdc0ee
--- /dev/null
+++ b/src/stream/test_stream_2peers.c
@@ -0,0 +1,560 @@
+/*
+     This file is part of GNUnet.
+     (C) 2011, 2012 Christian Grothoff (and other contributing authors)
+
+     GNUnet is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 3, or (at your
+     option) any later version.
+
+     GNUnet is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with GNUnet; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+     Boston, MA 02111-1307, USA.
+*/
+
+/**
+ * @file stream/test_stream_2peers.c
+ * @brief Stream API testing between 2 peers using testing API
+ * @author Sree Harsha Totakura
+ */
+
+#include <string.h>
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_mesh_service.h"
+#include "gnunet_stream_lib.h"
+#include "gnunet_testing_lib.h"
+
+#define VERBOSE 1
+
+/**
+ * Number of peers
+ */
+#define NUM_PEERS 2
+
+/**
+ * Structure for holding peer's sockets and IO Handles
+ */
+struct PeerData
+{
+  /**
+   * Peer's stream socket
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Peer's io write handle
+   */
+  struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
+
+  /**
+   * Peer's io read handle
+   */
+  struct GNUNET_STREAM_IOReadHandle *io_read_handle;
+
+  /**
+   * Peer's shutdown handle
+   */
+  struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
+
+  /**
+   * Our Peer id
+   */
+  struct GNUNET_PeerIdentity our_id;
+
+  /**
+   * Bytes the peer has written
+   */
+  unsigned int bytes_wrote;
+
+  /**
+   * Byte the peer has read
+   */
+  unsigned int bytes_read;
+};
+
+/**
+ * The current peer group
+ */
+static struct GNUNET_TESTING_PeerGroup *pg;
+
+/**
+ * Peer 1 daemon
+ */
+static struct GNUNET_TESTING_Daemon *d1;
+
+/**
+ * Peer 2 daemon
+ */
+static struct GNUNET_TESTING_Daemon *d2;
+
+static struct PeerData peer1;
+static struct PeerData peer2;
+static struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
+static struct GNUNET_CONFIGURATION_Handle *config;
+
+static GNUNET_SCHEDULER_TaskIdentifier abort_task;
+
+static char *data = "ABCD";
+static int result;
+
+static int writing_success;
+static int reading_success;
+
+
+/**
+ * Input processor
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+input_processor (void *cls,
+                 enum GNUNET_STREAM_Status status,
+                 const void *input_data,
+                 size_t size);
+
+/**
+ * Task for calling STREAM_read
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_read_task (void *cls,
+                  const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
+                                             GNUNET_TIME_relative_multiply
+                                             (GNUNET_TIME_UNIT_SECONDS, 5),
+                                             &input_processor,
+                                             peer);
+  GNUNET_assert (NULL != peer->io_read_handle);
+}
+
+/**
+ * The write completion function; called upon writing some data to stream or
+ * upon error
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param size the number of bytes read or written
+ */
+static void 
+write_completion (void *cls,
+                  enum GNUNET_STREAM_Status status,
+                  size_t size);
+
+
+/**
+ * Task for calling STREAM_write
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_write_task (void *cls,
+                   const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_write_handle = 
+    GNUNET_STREAM_write (peer->socket,
+                         (void *) data,
+                         strlen(data) - peer->bytes_wrote,
+                         GNUNET_TIME_relative_multiply
+                         (GNUNET_TIME_UNIT_SECONDS, 5),
+                         &write_completion,
+                         peer);
+ 
+  GNUNET_assert (NULL != peer->io_write_handle);
+ }
+
+
+/**
+ * Check whether peers successfully shut down.
+ */
+static void
+peergroup_shutdown_callback (void *cls, const char *emsg)
+{
+  if (emsg != NULL)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "Shutdown of peers failed!\n");
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "All peers successfully shut down!\n");
+  }
+  GNUNET_CONFIGURATION_destroy (config);
+}
+
+
+/**
+ * Close sockets and stop testing deamons nicely
+ */
+static void
+do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  if (NULL != peer1.socket)
+    GNUNET_STREAM_close (peer1.socket);
+  if (NULL != peer2.socket)
+    GNUNET_STREAM_close (peer2.socket);
+  if (NULL != peer2_listen_socket)
+    GNUNET_STREAM_listen_close (peer2_listen_socket);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: shutdown\n");
+  if (0 != abort_task)
+  {
+    GNUNET_SCHEDULER_cancel (abort_task);
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: Wait\n");
+
+  GNUNET_TESTING_daemons_stop (pg,
+                               GNUNET_TIME_relative_multiply
+                               (GNUNET_TIME_UNIT_SECONDS, 5),
+                               &peergroup_shutdown_callback,
+                               NULL);
+}
+
+
+/**
+ * Completion callback for shutdown
+ *
+ * @param cls the closure from GNUNET_STREAM_shutdown call
+ * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
+ *          SHUT_RDWR) 
+ */
+static void 
+shutdown_completion (void *cls,
+                     int operation)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "STREAM shutdown successful\n");
+  GNUNET_SCHEDULER_add_now (&do_close,
+                            cls);
+}
+
+
+
+/**
+ * Shutdown sockets gracefully
+ */
+static void
+do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  peer1.shutdown_handle = GNUNET_STREAM_shutdown (peer1.socket, 
+                                                  SHUT_RDWR,
+                                                  &shutdown_completion,
+                                                  cls);
+}
+
+
+/**
+ * Something went wrong and timed out. Kill everything and set error flag
+ */
+static void
+do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
+  result = GNUNET_SYSERR;
+  abort_task = 0;
+  do_close (cls, tc);  
+}
+
+
+/**
+ * The write completion function; called upon writing some data to stream or
+ * upon error
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param size the number of bytes read or written
+ */
+static void 
+write_completion (void *cls,
+                  enum GNUNET_STREAM_Status status,
+                  size_t size)
+{
+  struct PeerData *peer=cls;
+
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+  GNUNET_assert (size <= strlen (data));
+  peer->bytes_wrote += size;
+
+  if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
+    {
+      GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
+    }
+  else
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Writing completed\n");
+
+      if (&peer1 == peer)   /* Peer1 has finished writing; should read now */
+        {
+          peer->bytes_read = 0;
+          GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+        }
+      else
+        {
+          writing_success = GNUNET_YES;
+          if (GNUNET_YES == reading_success)
+            GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+        }
+    }
+}
+
+
+/**
+ * Function executed after stream has been established
+ *
+ * @param cls the closure from GNUNET_STREAM_open
+ * @param socket socket to use to communicate with the other side (read/write)
+ */
+static void 
+stream_open_cb (void *cls,
+                struct GNUNET_STREAM_Socket *socket)
+{
+  struct PeerData *peer=cls;
+  
+  GNUNET_assert (&peer1 == peer);
+  GNUNET_assert (socket == peer1.socket);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s: Stream established from peer1\n",
+              GNUNET_i2s (&peer1.our_id));
+  peer->bytes_wrote = 0;
+  GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
+}
+
+
+/**
+ * Input processor
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+input_processor (void *cls,
+                 enum GNUNET_STREAM_Status status,
+                 const void *input_data,
+                 size_t size)
+{
+  struct PeerData *peer;
+
+  peer = (struct PeerData *) cls;
+
+  if (GNUNET_STREAM_TIMEOUT == status)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Read operation timedout - reading again!\n");
+      GNUNET_assert (0 == size);
+      GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+      return 0;
+    }
+
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+  GNUNET_assert (size <= strlen (data));
+  GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read, 
+                               (const char *) input_data,
+                               size));
+  peer->bytes_read += size;
+  
+  if (peer->bytes_read < strlen (data))
+    {
+      GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+    }
+  else 
+    {
+      if (&peer2 == peer)    /* Peer2 has completed reading; should write */
+        {
+          peer->bytes_wrote = 0;
+          GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
+        }
+      else                      /* Peer1 has completed reading. End of tests */
+        {
+          reading_success = GNUNET_YES;
+          if (GNUNET_YES == writing_success)
+            GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+        }
+    }
+  return size;
+}
+
+  
+/**
+ * Functions of this type are called upon new stream connection from other peers
+ *
+ * @param cls the closure from GNUNET_STREAM_listen
+ * @param socket the socket representing the stream
+ * @param initiator the identity of the peer who wants to establish a stream
+ *            with us
+ * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
+ *             stream (the socket will be invalid after the call)
+ */
+static int
+stream_listen_cb (void *cls,
+                  struct GNUNET_STREAM_Socket *socket,
+                  const struct GNUNET_PeerIdentity *initiator)
+{
+  GNUNET_assert (NULL != socket);
+  GNUNET_assert (NULL != initiator);
+  GNUNET_assert (socket != peer1.socket);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s: Peer connected: %s\n",
+              GNUNET_i2s (&peer2.our_id),
+              GNUNET_i2s(initiator));
+
+  peer2.socket = socket;
+  peer2.bytes_read = 0;
+  GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Callback to be called when testing peer group is ready
+ *
+ * @param cls NULL
+ * @param emsg NULL on success
+ */
+void
+peergroup_ready (void *cls, const char *emsg)
+{
+  if (NULL != emsg)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Starting peer group failed: %s\n", emsg);
+      return;
+    }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Peer group is now ready\n");
+  
+  GNUNET_assert (2 == GNUNET_TESTING_daemons_running (pg));
+  
+  d1 = GNUNET_TESTING_daemon_get (pg, 0);
+  GNUNET_assert (NULL != d1);
+  
+  d2 = GNUNET_TESTING_daemon_get (pg, 1);
+  GNUNET_assert (NULL != d2);
+
+  GNUNET_TESTING_get_peer_identity (d1->cfg,
+                                    &peer1.our_id);
+  GNUNET_TESTING_get_peer_identity (d2->cfg,
+                                    &peer2.our_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s : %s\n",
+              GNUNET_i2s (&peer1.our_id),
+              GNUNET_i2s (&d1->id));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s : %s\n",
+              GNUNET_i2s (&peer2.our_id),
+              GNUNET_i2s (&d2->id));
+
+  peer2_listen_socket = GNUNET_STREAM_listen (d2->cfg,
+                                              10, /* App port */
+                                              &stream_listen_cb,
+                                              NULL);
+  GNUNET_assert (NULL != peer2_listen_socket);
+
+  /* Connect to stream library */
+  peer1.socket = GNUNET_STREAM_open (d1->cfg,
+                                     &d2->id,         /* Null for local peer? */
+                                     10,           /* App port */
+                                     &stream_open_cb,
+                                     &peer1,
+				     GNUNET_STREAM_OPTION_END);
+  GNUNET_assert (NULL != peer1.socket);
+}
+
+
+/**
+ * Initialize framework and start test
+ */
+static void
+run (void *cls, char *const *args, const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  struct GNUNET_TESTING_Host *hosts; /* FIXME: free hosts (DLL) */
+
+  GNUNET_log_setup ("test_stream_2peers",
+                    "DEBUG",
+                    NULL);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Starting test\n");
+  /* Duplicate the configuration */
+  config = GNUNET_CONFIGURATION_dup (cfg);
+
+  hosts = GNUNET_TESTING_hosts_load (config);
+  
+  pg = GNUNET_TESTING_peergroup_start (config,
+                                       2,
+                                       GNUNET_TIME_relative_multiply
+                                       (GNUNET_TIME_UNIT_SECONDS, 3),
+                                       NULL,
+                                       &peergroup_ready,
+                                       NULL,
+                                       hosts);
+  GNUNET_assert (NULL != pg);
+                                       
+  abort_task =
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+                                  (GNUNET_TIME_UNIT_SECONDS, 40), &do_abort,
+                                  NULL);
+}
+
+/**
+ * Main function
+ */
+int main (int argc, char **argv)
+{
+  int ret;
+
+  char *argv2[] = { "test-stream-2peers",
+                    "-L", "DEBUG",
+                    "-c", "test_stream_local.conf",
+                    NULL};
+  
+  struct GNUNET_GETOPT_CommandLineOption options[] = {
+    GNUNET_GETOPT_OPTION_END
+  };
+
+  ret =
+      GNUNET_PROGRAM_run ((sizeof (argv2) / sizeof (char *)) - 1, argv2,
+                          "test-stream-2peers", "nohelp", options, &run, NULL);
+
+  if (GNUNET_OK != ret)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "run failed with error code %d\n",
+                ret);
+    return 1;
+  }
+  if (GNUNET_SYSERR == result)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test failed\n");
+    return 1;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test ok\n");
+  return 0;
+}
diff --git a/src/stream/test_stream_2peers_halfclose.c b/src/stream/test_stream_2peers_halfclose.c
new file mode 100644
index 0000000..7997c20
--- /dev/null
+++ b/src/stream/test_stream_2peers_halfclose.c
@@ -0,0 +1,786 @@
+/*
+     This file is part of GNUnet.
+     (C) 2011, 2012 Christian Grothoff (and other contributing authors)
+
+     GNUnet is free software; you can redistribute it and/or modify
+     it under the terms of the GNU General Public License as published
+     by the Free Software Foundation; either version 3, or (at your
+     option) any later version.
+
+     GNUnet is distributed in the hope that it will be useful, but
+     WITHOUT ANY WARRANTY; without even the implied warranty of
+     MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+     General Public License for more details.
+
+     You should have received a copy of the GNU General Public License
+     along with GNUnet; see the file COPYING.  If not, write to the
+     Free Software Foundation, Inc., 59 Temple Place - Suite 330,
+     Boston, MA 02111-1307, USA.
+*/
+
+/**
+ * @file stream/test_stream_2peers_halfclose.c
+ * @brief Testcases for Stream API halfclosed connections between 2 peers
+ * @author Sree Harsha Totakura
+ */
+
+#include <string.h>
+
+#include "platform.h"
+#include "gnunet_util_lib.h"
+#include "gnunet_mesh_service.h"
+#include "gnunet_stream_lib.h"
+#include "gnunet_testing_lib.h"
+#include "gnunet_scheduler_lib.h"
+
+#define VERBOSE 1
+
+/**
+ * Number of peers
+ */
+#define NUM_PEERS 2
+
+/**
+ * Structure for holding peer's sockets and IO Handles
+ */
+struct PeerData
+{
+  /**
+   * Peer's stream socket
+   */
+  struct GNUNET_STREAM_Socket *socket;
+
+  /**
+   * Peer's io write handle
+   */
+  struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
+
+  /**
+   * Peer's io read handle
+   */
+  struct GNUNET_STREAM_IOReadHandle *io_read_handle;
+
+  /**
+   * Peer's shutdown handle
+   */
+  struct GNUNET_STREAM_ShutdownHandle *shutdown_handle;
+
+  /**
+   * Our Peer id
+   */
+  struct GNUNET_PeerIdentity our_id;
+
+  /**
+   * Bytes the peer has written
+   */
+  unsigned int bytes_wrote;
+
+  /**
+   * Byte the peer has read
+   */
+  unsigned int bytes_read;
+
+  /**
+   * GNUNET_YES if the peer has successfully completed the current test
+   */
+  unsigned int test_ok;
+
+  /**
+   * The shutdown operation that has to be used by the stream_shutdown_task
+   */
+  int shutdown_operation;
+};
+
+/**
+ * The current peer group
+ */
+static struct GNUNET_TESTING_PeerGroup *pg;
+
+/**
+ * Peer 1 daemon
+ */
+static struct GNUNET_TESTING_Daemon *d1;
+
+/**
+ * Peer 2 daemon
+ */
+static struct GNUNET_TESTING_Daemon *d2;
+
+
+/**
+ * Peer1 writes first and then calls for SHUT_WR
+ * Peer2 reads first and then calls for SHUT_RD
+ * Attempt to write again by Peer1 should be rejected
+ * Attempt to read again by Peer2 should be rejected
+ * Peer1 then reads from Peer2 which writes
+ */
+static struct PeerData peer1;
+static struct PeerData peer2;
+static struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
+static struct GNUNET_CONFIGURATION_Handle *config;
+
+static GNUNET_SCHEDULER_TaskIdentifier abort_task;
+static GNUNET_SCHEDULER_TaskIdentifier read_task;
+
+static char *data = "ABCD";
+static int result;
+
+/**
+ * Enumeration for various tests that are to be passed in the same order as
+ * below
+ */
+enum Test
+  {
+    /**
+     * Peer1 writing; Peer2 reading
+     */
+    PEER1_WRITE,
+
+    /**
+     * Peer1 write shutdown; Peer2 should get an error when it tries to read;
+     */
+    PEER1_WRITE_SHUTDOWN,
+
+    /**
+     * Peer1 reads; Peer2 writes (connection is halfclosed)
+     */
+    PEER1_HALFCLOSE_READ,
+
+    /**
+     * Peer1 attempts to write; Should fail with stream already shutdown error
+     */
+    PEER1_HALFCLOSE_WRITE_FAIL,
+
+    /**
+     * Peer1 read shutdown; Peer2 should get stream shutdown error during write
+     */
+    PEER1_READ_SHUTDOWN,
+
+    /**
+     * All tests successfully finished
+     */
+    SUCCESS
+  };
+
+/**
+ * Current running test
+ */
+enum Test current_test;
+
+/**
+ * Input processor
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+input_processor (void *cls,
+                 enum GNUNET_STREAM_Status status,
+                 const void *input_data,
+                 size_t size);
+
+
+/**
+ * The transition function; responsible for the transitions among tests
+ */
+static void
+transition();
+
+
+/**
+ * Task for calling STREAM_read
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_read_task (void *cls,
+                  const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
+                                             GNUNET_TIME_relative_multiply
+                                             (GNUNET_TIME_UNIT_SECONDS, 5),
+                                             &input_processor,
+                                             cls);
+  switch (current_test)
+    {
+    case PEER1_WRITE_SHUTDOWN:
+      GNUNET_assert (&peer2 == peer);
+      GNUNET_assert (NULL == peer->io_read_handle);
+      transition ();            /* to PEER1_HALFCLOSE_READ */
+      break;
+    default:
+      GNUNET_assert (NULL != peer->io_read_handle);
+    }
+}
+
+
+/**
+ * The write completion function; called upon writing some data to stream or
+ * upon error
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param size the number of bytes read or written
+ */
+static void 
+write_completion (void *cls,
+                  enum GNUNET_STREAM_Status status,
+                  size_t size);
+
+
+/**
+ * Task for calling STREAM_write
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_write_task (void *cls,
+                   const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_write_handle = 
+    GNUNET_STREAM_write (peer->socket,
+                         (void *) data,
+                         strlen(data) - peer->bytes_wrote,
+                         GNUNET_TIME_relative_multiply
+                         (GNUNET_TIME_UNIT_SECONDS, 5),
+                         &write_completion,
+                         peer);
+  switch (current_test)
+    {
+    case PEER1_HALFCLOSE_WRITE_FAIL:
+      GNUNET_assert (&peer1 == peer);
+      GNUNET_assert (NULL == peer->io_write_handle);
+      transition();             /* To PEER1_READ_SHUTDOWN */
+      break;
+    case PEER1_READ_SHUTDOWN:
+      GNUNET_assert (&peer2 == peer);
+      GNUNET_assert (NULL == peer->io_write_handle);
+      transition ();            /* To SUCCESS */
+      break;
+    default:
+        GNUNET_assert (NULL != peer->io_write_handle);
+    }
+}
+
+
+/**
+ * Check whether peers successfully shut down.
+ */
+static void
+peergroup_shutdown_callback (void *cls, const char *emsg)
+{
+  if (emsg != NULL)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING,
+                "Shutdown of peers failed!\n");
+  }
+  else
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_INFO,
+                "All peers successfully shut down!\n");
+  }
+  GNUNET_CONFIGURATION_destroy (config);
+}
+
+
+/**
+ * Close sockets and stop testing deamons nicely
+ */
+static void
+do_close (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  if (NULL != peer1.socket)
+    GNUNET_STREAM_close (peer1.socket);
+  if (NULL != peer2.socket)
+    GNUNET_STREAM_close (peer2.socket);
+  if (NULL != peer2_listen_socket)
+    GNUNET_STREAM_listen_close (peer2_listen_socket);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: shutdown\n");
+  if (0 != abort_task)
+  {
+    GNUNET_SCHEDULER_cancel (abort_task);
+  }
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: Wait\n");
+
+  GNUNET_TESTING_daemons_stop (pg,
+                               GNUNET_TIME_relative_multiply
+                               (GNUNET_TIME_UNIT_SECONDS, 5),
+                               &peergroup_shutdown_callback,
+                               NULL);
+}
+
+
+/**
+ * Completion callback for shutdown
+ *
+ * @param cls the closure from GNUNET_STREAM_shutdown call
+ * @param operation the operation that was shutdown (SHUT_RD, SHUT_WR,
+ *          SHUT_RDWR) 
+ */
+void 
+shutdown_completion (void *cls,
+                     int operation)
+{
+  switch (current_test)
+    {
+    case PEER1_WRITE:
+      GNUNET_assert (0);
+    case PEER1_WRITE_SHUTDOWN:
+      GNUNET_assert (cls == &peer1);
+      GNUNET_assert (SHUT_WR == operation);
+      peer1.test_ok = GNUNET_YES;
+      /* Peer2 should read with error */
+      peer2.bytes_read = 0;
+      GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
+      break;
+    case PEER1_READ_SHUTDOWN:
+      peer1.test_ok = GNUNET_YES;
+      peer2.bytes_wrote = 0;
+      GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
+      break;
+    case PEER1_HALFCLOSE_READ:
+    case PEER1_HALFCLOSE_WRITE_FAIL:
+    case SUCCESS:
+      GNUNET_assert (0);        /* We shouldn't reach here */
+    }
+}
+
+
+/**
+ * Task for calling STREAM_shutdown
+ *
+ * @param cls the peer entity
+ * @param tc the TaskContext
+ */
+static void
+stream_shutdown_task (void *cls,
+                      const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+
+  peer->shutdown_handle = GNUNET_STREAM_shutdown (peer->socket,
+                                                  peer->shutdown_operation,
+                                                  &shutdown_completion,
+                                                  peer);
+  GNUNET_assert (NULL != peer->shutdown_handle);
+}
+
+
+/**
+ * Something went wrong and timed out. Kill everything and set error flag
+ */
+static void
+do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: ABORT\n");
+  if (0 != read_task)
+    {
+      GNUNET_SCHEDULER_cancel (read_task);
+    }
+  result = GNUNET_SYSERR;
+  abort_task = 0;
+  do_close (cls, tc);  
+}
+
+
+/**
+ * The transition function; responsible for the transitions among tests
+ */
+static void
+transition()
+{
+  if ((GNUNET_YES == peer1.test_ok) && (GNUNET_YES == peer2.test_ok))
+    {
+      peer1.test_ok = GNUNET_NO;
+      peer2.test_ok = GNUNET_NO;
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "TEST %d SUCCESSFULL\n", current_test);
+      switch (current_test)
+        {
+        case PEER1_WRITE:
+          current_test = PEER1_WRITE_SHUTDOWN;
+          /* Peer1 should shutdown writing */
+          peer1.shutdown_operation = SHUT_WR;
+          GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
+          break;
+        case PEER1_WRITE_SHUTDOWN:
+          current_test = PEER1_HALFCLOSE_READ;
+          /* Peer2 should be able to write successfully */
+          peer2.bytes_wrote = 0;
+          GNUNET_SCHEDULER_add_now (&stream_write_task, &peer2);
+          
+          /* Peer1 should be able to read successfully */
+          peer1.bytes_read = 0;
+          GNUNET_SCHEDULER_add_now (&stream_read_task, &peer1);
+          break;
+        case PEER1_HALFCLOSE_READ:
+          current_test = PEER1_HALFCLOSE_WRITE_FAIL;
+          peer1.bytes_wrote = 0;
+          peer2.bytes_read = 0;
+          peer2.test_ok = GNUNET_YES;
+          GNUNET_SCHEDULER_add_now (&stream_write_task, &peer1);
+          break;
+        case PEER1_HALFCLOSE_WRITE_FAIL:
+          current_test = PEER1_READ_SHUTDOWN;
+          peer1.shutdown_operation = SHUT_RD;
+          GNUNET_SCHEDULER_add_now (&stream_shutdown_task, &peer1);
+          break;
+        case PEER1_READ_SHUTDOWN:
+          current_test = SUCCESS;
+          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                      "All tests successful\n");
+          GNUNET_SCHEDULER_add_now (&do_close, NULL);
+          break;
+        case SUCCESS:
+          GNUNET_assert (0);    /* We shouldn't reach here */
+          
+        }
+    }
+}
+
+/**
+ * The write completion function; called upon writing some data to stream or
+ * upon error
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param size the number of bytes read or written
+ */
+static void 
+write_completion (void *cls,
+                  enum GNUNET_STREAM_Status status,
+                  size_t size)
+{
+  struct PeerData *peer = cls;
+
+  switch (current_test)
+    {
+    case PEER1_WRITE:
+    case PEER1_HALFCLOSE_READ:
+
+    GNUNET_assert (GNUNET_STREAM_OK == status);
+    GNUNET_assert (size <= strlen (data));
+    peer->bytes_wrote += size;
+
+    if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
+      {
+        GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
+      }
+    else
+      {
+        GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                    "Writing completed\n");
+
+        if (&peer1 == peer)
+          {
+            peer1.test_ok = GNUNET_YES;
+            transition ();       /* to PEER1_WRITE_SHUTDOWN */
+          }
+        else            /* This will happen during PEER1_HALFCLOSE_READ */
+          {
+            peer2.test_ok = GNUNET_YES;
+            transition ();      /* to PEER1_HALFCLOSE_WRITE_FAIL */
+          }
+      }
+    break;
+    case PEER1_HALFCLOSE_WRITE_FAIL:
+      GNUNET_assert (peer == &peer1);
+      GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
+      GNUNET_assert (0 == size);
+      peer1.test_ok = GNUNET_YES;
+      break;
+    case PEER1_READ_SHUTDOWN:
+      GNUNET_assert (peer == &peer2);
+      GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
+      GNUNET_assert (0 == size);
+      peer2.test_ok = GNUNET_YES;
+      break;
+    case PEER1_WRITE_SHUTDOWN:
+    case SUCCESS:
+      GNUNET_assert (0);        /* We shouldn't reach here */
+    } 
+}
+
+
+/**
+ * Function executed after stream has been established
+ *
+ * @param cls the closure from GNUNET_STREAM_open
+ * @param socket socket to use to communicate with the other side (read/write)
+ */
+static void 
+stream_open_cb (void *cls,
+                struct GNUNET_STREAM_Socket *socket)
+{
+  struct PeerData *peer;
+
+  GNUNET_assert (socket == peer1.socket);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s: Stream established from peer1\n",
+              GNUNET_i2s (&peer1.our_id));
+  peer = (struct PeerData *) cls;
+  peer->bytes_wrote = 0;
+  GNUNET_assert (socket == peer1.socket);
+  GNUNET_assert (socket == peer->socket);
+  peer1.test_ok = GNUNET_NO;
+  peer2.test_ok = GNUNET_NO;
+  current_test = PEER1_WRITE;
+  GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
+}
+
+
+/**
+ * Input processor
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+input_processor (void *cls,
+                 enum GNUNET_STREAM_Status status,
+                 const void *input_data,
+                 size_t size)
+{
+  struct PeerData *peer;
+
+  peer = (struct PeerData *) cls;
+
+  switch (current_test)
+    {
+    case PEER1_WRITE:
+    case PEER1_HALFCLOSE_READ:
+      if (GNUNET_STREAM_TIMEOUT == status)
+        {
+          GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                      "Read operation timedout - reading again!\n");
+          GNUNET_assert (0 == size);
+          GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+          return 0;
+        }
+
+      GNUNET_assert (GNUNET_STREAM_OK == status);
+      GNUNET_assert (size <= strlen (data));
+      GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read,
+                                   (const char *) input_data,
+                                   size));
+      peer->bytes_read += size;
+  
+      if (peer->bytes_read < strlen (data))
+        {
+          GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+        }
+      else  
+        {
+          if (&peer2 == peer) /* Peer2 has completed reading; should write */
+            {
+              peer2.test_ok = GNUNET_YES;
+              transition ();    /* Transition to PEER1_WRITE_SHUTDOWN */
+            }
+          else         /* Peer1 has completed reading. End of tests */
+            {
+              peer1.test_ok = GNUNET_YES;
+              transition ();    /* to PEER1_HALFCLOSE_WRITE_FAIL */
+            }
+        }
+      break;
+    case PEER1_WRITE_SHUTDOWN:
+      GNUNET_assert (GNUNET_STREAM_SHUTDOWN == status);
+      peer2.test_ok = GNUNET_YES;
+      break;
+    case PEER1_HALFCLOSE_WRITE_FAIL:
+    case PEER1_READ_SHUTDOWN:
+    case SUCCESS:
+      GNUNET_assert (0);        /* We shouldn't reach here */
+    }
+  
+  return size;
+}
+
+  
+/**
+ * Scheduler call back; to be executed when a new stream is connected
+ * Called from listen connect for peer2
+ */
+static void
+stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  read_task = GNUNET_SCHEDULER_NO_TASK;
+  GNUNET_assert (NULL != cls);
+  peer2.bytes_read = 0;
+  GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
+}
+
+
+/**
+ * Functions of this type are called upon new stream connection from other peers
+ *
+ * @param cls the closure from GNUNET_STREAM_listen
+ * @param socket the socket representing the stream
+ * @param initiator the identity of the peer who wants to establish a stream
+ *            with us
+ * @return GNUNET_OK to keep the socket open, GNUNET_SYSERR to close the
+ *             stream (the socket will be invalid after the call)
+ */
+static int
+stream_listen_cb (void *cls,
+                  struct GNUNET_STREAM_Socket *socket,
+                  const struct GNUNET_PeerIdentity *initiator)
+{
+  GNUNET_assert (NULL != socket);
+  GNUNET_assert (NULL != initiator);
+  GNUNET_assert (socket != peer1.socket);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s: Peer connected: %s\n",
+              GNUNET_i2s (&peer2.our_id),
+              GNUNET_i2s(initiator));
+
+  peer2.socket = socket;
+  /* FIXME: reading should be done right now instead of a scheduled call */
+  read_task = GNUNET_SCHEDULER_add_now (&stream_read, (void *) socket);
+  return GNUNET_OK;
+}
+
+
+/**
+ * Callback to be called when testing peer group is ready
+ *
+ * @param cls NULL
+ * @param emsg NULL on success
+ */
+void
+peergroup_ready (void *cls, const char *emsg)
+{
+  if (NULL != emsg)
+    {
+      GNUNET_log (GNUNET_ERROR_TYPE_ERROR,
+                  "Starting peer group failed: %s\n", emsg);
+      return;
+    }
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Peer group is now ready\n");
+  
+  GNUNET_assert (2 == GNUNET_TESTING_daemons_running (pg));
+  
+  d1 = GNUNET_TESTING_daemon_get (pg, 0);
+  GNUNET_assert (NULL != d1);
+  
+  d2 = GNUNET_TESTING_daemon_get (pg, 1);
+  GNUNET_assert (NULL != d2);
+
+  GNUNET_TESTING_get_peer_identity (d1->cfg,
+                                    &peer1.our_id);
+  GNUNET_TESTING_get_peer_identity (d2->cfg,
+                                    &peer2.our_id);
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s : %s\n",
+              GNUNET_i2s (&peer1.our_id),
+              GNUNET_i2s (&d1->id));
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "%s : %s\n",
+              GNUNET_i2s (&peer2.our_id),
+              GNUNET_i2s (&d2->id));
+
+  peer2_listen_socket = GNUNET_STREAM_listen (d2->cfg,
+                                              10, /* App port */
+                                              &stream_listen_cb,
+                                              NULL);
+  GNUNET_assert (NULL != peer2_listen_socket);
+
+  /* Connect to stream library */
+  peer1.socket = GNUNET_STREAM_open (d1->cfg,
+                                     &d2->id,         /* Null for local peer? */
+                                     10,           /* App port */
+                                     &stream_open_cb,
+                                     &peer1,
+				     GNUNET_STREAM_OPTION_END);
+  GNUNET_assert (NULL != peer1.socket);
+}
+
+
+/**
+ * Initialize framework and start test
+ */
+static void
+run (void *cls, char *const *args, const char *cfgfile,
+     const struct GNUNET_CONFIGURATION_Handle *cfg)
+{
+  struct GNUNET_TESTING_Host *hosts; /* FIXME: free hosts (DLL) */
+
+  /* GNUNET_log_setup ("test_stream_local", */
+  /*                   "DEBUG", */
+  /*                   NULL); */
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Starting test\n");
+  /* Duplicate the configuration */
+  config = GNUNET_CONFIGURATION_dup (cfg);
+
+  hosts = GNUNET_TESTING_hosts_load (config);
+  
+  pg = GNUNET_TESTING_peergroup_start (config,
+                                       2,
+                                       GNUNET_TIME_relative_multiply
+                                       (GNUNET_TIME_UNIT_SECONDS, 3),
+                                       NULL,
+                                       &peergroup_ready,
+                                       NULL,
+                                       hosts);
+  GNUNET_assert (NULL != pg);
+                                       
+  abort_task =
+    GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
+                                  (GNUNET_TIME_UNIT_SECONDS, 40), &do_abort,
+                                  NULL);
+}
+
+/**
+ * Main function
+ */
+int main (int argc, char **argv)
+{
+  int ret;
+
+  char *argv2[] = { "test-stream-2peers-halfclose",
+                    "-L", "DEBUG",
+                    "-c", "test_stream_local.conf",
+                    NULL};
+  
+  struct GNUNET_GETOPT_CommandLineOption options[] = {
+    GNUNET_GETOPT_OPTION_END
+  };
+
+  ret =
+      GNUNET_PROGRAM_run ((sizeof (argv2) / sizeof (char *)) - 1, argv2,
+                          "test-stream-2peers-halfclose", "nohelp", options, &run, NULL);
+
+  if (GNUNET_OK != ret)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "run failed with error code %d\n",
+                ret);
+    return 1;
+  }
+  if (GNUNET_SYSERR == result)
+  {
+    GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test failed\n");
+    return 1;
+  }
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test ok\n");
+  return 0;
+}
diff --git a/src/stream/test_stream_local.c b/src/stream/test_stream_local.c
index 4da1258..c9fab84 100644
--- a/src/stream/test_stream_local.c
+++ b/src/stream/test_stream_local.c
@@ -30,6 +30,7 @@
 #include "gnunet_util_lib.h"
 #include "gnunet_mesh_service.h"
 #include "gnunet_stream_lib.h"
+#include "gnunet_testing_lib.h"
 
 #define VERBOSE 1
 
@@ -44,9 +45,14 @@ struct PeerData
   struct GNUNET_STREAM_Socket *socket;
 
   /**
-   * Peer's io handle
+   * Peer's io write handle
    */
-  struct GNUNET_STREAM_IOHandle *io_handle;
+  struct GNUNET_STREAM_IOWriteHandle *io_write_handle;
+
+  /**
+   * Peer's io read handle
+   */
+  struct GNUNET_STREAM_IOReadHandle *io_read_handle;
 
   /**
    * Bytes the peer has written
@@ -63,14 +69,92 @@ static struct GNUNET_OS_Process *arm_pid;
 static struct PeerData peer1;
 static struct PeerData peer2;
 static struct GNUNET_STREAM_ListenSocket *peer2_listen_socket;
+static struct GNUNET_CONFIGURATION_Handle *config_peer1;
+static struct GNUNET_CONFIGURATION_Handle *config_peer2;
 
 static GNUNET_SCHEDULER_TaskIdentifier abort_task;
 static GNUNET_SCHEDULER_TaskIdentifier test_task;
-static GNUNET_SCHEDULER_TaskIdentifier read_task;
 
 static char *data = "ABCD";
 static int result;
 
+static int writing_success;
+static int reading_success;
+
+/**
+ * Input processor
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param data traffic from the other side
+ * @param size the number of bytes available in data read 
+ * @return number of bytes of processed from 'data' (any data remaining should be
+ *         given to the next time the read processor is called).
+ */
+static size_t
+input_processor (void *cls,
+                 enum GNUNET_STREAM_Status status,
+                 const void *input_data,
+                 size_t size);
+
+/**
+ * Task for calling STREAM_read
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_read_task (void *cls,
+                  const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_read_handle = GNUNET_STREAM_read (peer->socket,
+                                             GNUNET_TIME_relative_multiply
+                                             (GNUNET_TIME_UNIT_SECONDS, 5),
+                                             &input_processor,
+                                             peer);
+  GNUNET_assert (NULL != peer->io_read_handle);
+}
+
+/**
+ * The write completion function; called upon writing some data to stream or
+ * upon error
+ *
+ * @param cls the closure from GNUNET_STREAM_write/read
+ * @param status the status of the stream at the time this function is called
+ * @param size the number of bytes read or written
+ */
+static void 
+write_completion (void *cls,
+                  enum GNUNET_STREAM_Status status,
+                  size_t size);
+
+
+/**
+ * Task for calling STREAM_write
+ *
+ * @param cls the peer data entity
+ * @param tc the task context
+ */
+static void
+stream_write_task (void *cls,
+                   const struct GNUNET_SCHEDULER_TaskContext *tc)
+{
+  struct PeerData *peer = cls;
+  
+  peer->io_write_handle = 
+    GNUNET_STREAM_write (peer->socket,
+                         (void *) data,
+                         strlen(data) - peer->bytes_wrote,
+                         GNUNET_TIME_relative_multiply
+                         (GNUNET_TIME_UNIT_SECONDS, 5),
+                         &write_completion,
+                         peer);
+ 
+  GNUNET_assert (NULL != peer->io_write_handle);
+ }
+
 /**
  * Shutdown nicely
  */
@@ -78,7 +162,10 @@ static void
 do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
   GNUNET_STREAM_close (peer1.socket);
-  GNUNET_STREAM_close (peer2.socket);
+  if (NULL != peer2.socket)
+    GNUNET_STREAM_close (peer2.socket);
+  if (NULL != peer2_listen_socket)
+    GNUNET_STREAM_listen_close (peer2_listen_socket);
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: shutdown\n");
   if (0 != abort_task)
   {
@@ -90,8 +177,11 @@ do_shutdown (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
     GNUNET_log_strerror (GNUNET_ERROR_TYPE_WARNING, "kill");
   }
   GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test: Wait\n");
+  /* Free the duplicated configuration */
+  GNUNET_CONFIGURATION_destroy (config_peer1);
+  GNUNET_CONFIGURATION_destroy (config_peer2);
   GNUNET_assert (GNUNET_OK == GNUNET_OS_process_wait (arm_pid));
-  GNUNET_OS_process_close (arm_pid);
+  GNUNET_OS_process_destroy (arm_pid);
 }
 
 
@@ -106,10 +196,7 @@ do_abort (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
   {
     GNUNET_SCHEDULER_cancel (test_task);
   }
-  if (0 != read_task)
-    {
-      GNUNET_SCHEDULER_cancel (read_task);
-    }
+
   result = GNUNET_SYSERR;
   abort_task = 0;
   do_shutdown (cls, tc);
@@ -129,35 +216,31 @@ write_completion (void *cls,
                   enum GNUNET_STREAM_Status status,
                   size_t size)
 {
-  struct PeerData *peer;
+  struct PeerData *peer=cls;
 
-  peer = (struct PeerData *) cls;
   GNUNET_assert (GNUNET_STREAM_OK == status);
-  GNUNET_assert (size < strlen (data));
+  GNUNET_assert (size <= strlen (data));
   peer->bytes_wrote += size;
 
   if (peer->bytes_wrote < strlen(data)) /* Have more data to send */
     {
-      peer->io_handle = GNUNET_STREAM_write (peer->socket,
-                                             (void *) data,
-                                             strlen(data) - peer->bytes_wrote,
-                                             GNUNET_TIME_relative_multiply
-                                             (GNUNET_TIME_UNIT_SECONDS, 5),
-                                             &write_completion,
-                                             cls);
-      GNUNET_assert (NULL != peer->io_handle);
+      GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
     }
   else
     {
+      GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+                  "Writing completed\n");
+
       if (&peer1 == peer)   /* Peer1 has finished writing; should read now */
         {
-          peer->io_handle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *)
-                                                peer->socket,
-                                                GNUNET_TIME_relative_multiply
-                                                (GNUNET_TIME_UNIT_SECONDS, 5),
-                                                &input_processor,
-                                                cls);
-          GNUNET_assert (NULL!=peer->io_handle);
+          peer->bytes_read = 0;
+          GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
+        }
+      else
+        {
+          writing_success = GNUNET_YES;
+          if (GNUNET_YES == reading_success)
+            GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
         }
     }
 }
@@ -173,20 +256,15 @@ static void
 stream_open_cb (void *cls,
                 struct GNUNET_STREAM_Socket *socket)
 {
-  struct PeerData *peer;
+  struct PeerData *peer=cls;
 
-  peer = (struct PeerData *) cls;
-  peer->bytes_wrote = 0;
+  GNUNET_assert (&peer1 == peer);
   GNUNET_assert (socket == peer1.socket);
   GNUNET_assert (socket == peer->socket);
-  peer->io_handle = GNUNET_STREAM_write (peer->socket, /* socket */
-                                         (void *) data, /* data */
-                                         strlen(data),
-                                         GNUNET_TIME_relative_multiply
-                                         (GNUNET_TIME_UNIT_SECONDS, 5),
-                                         &write_completion,
-                                         cls);
-  GNUNET_assert (NULL != peer->io_handle);
+
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "Stream established from peer1\n");
+  peer->bytes_wrote = 0;
+  GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
 }
 
 
@@ -206,44 +284,31 @@ input_processor (void *cls,
                  const void *input_data,
                  size_t size)
 {
-  struct PeerData *peer;
+  struct PeerData *peer = cls;
 
-  peer = (struct PeerData *) cls;
-
-  GNUNET_assert (GNUNET_STERAM_OK == status);
-  GNUNET_assert (size < strlen (data));
-  GNUNET_assert (strncmp ((const char *) data + peer->bytes_read, 
-                          (const char *) input_data,
-                          size));
+  GNUNET_assert (GNUNET_STREAM_OK == status);
+  GNUNET_assert (size <= strlen (data));
+  GNUNET_assert (0 == strncmp ((const char *) data + peer->bytes_read, 
+                               (const char *) input_data,
+                               size));
   peer->bytes_read += size;
   
   if (peer->bytes_read < strlen (data))
     {
-      peer->io_handle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *)
-                                            peer->socket,
-                                            GNUNET_TIME_relative_multiply
-                                            (GNUNET_TIME_UNIT_SECONDS, 5),
-                                            &input_processor,
-                                            cls);
-      GNUNET_assert (NULL != peer->io_handle);
+      GNUNET_SCHEDULER_add_now (&stream_read_task, peer);
     }
   else 
     {
       if (&peer2 == peer)    /* Peer2 has completed reading; should write */
         {
           peer->bytes_wrote = 0;
-          peer->io_handle = GNUNET_STREAM_write ((struct GNUNET_STREAM_Socket *)
-                                                 peer->socket,
-                                                 (void *) data,
-                                                 strlen(data),
-                                                 GNUNET_TIME_relative_multiply
-                                                 (GNUNET_TIME_UNIT_SECONDS, 5),
-                                                 &write_completion,
-                                                 cls);
+          GNUNET_SCHEDULER_add_now (&stream_write_task, peer);
         }
       else                      /* Peer1 has completed reading. End of tests */
         {
-          GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
+          reading_success = GNUNET_YES;
+          if (GNUNET_YES == writing_success)
+            GNUNET_SCHEDULER_add_now (&do_shutdown, NULL);
         }
     } 
   return size;
@@ -251,29 +316,9 @@ input_processor (void *cls,
 
   
 /**
- * Scheduler call back; to be executed when a new stream is connected
- * Called from listen connect for peer2
- */
-static void
-stream_read (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
-{
-  read_task = GNUNET_SCHEDULER_NO_TASK;
-  GNUNET_assert (NULL != cls);
-  peer2.bytes_read = 0;
-  GNUNET_STREAM_listen_close (peer2_listen_socket); /* Close listen socket */
-  peer2.io_handle = GNUNET_STREAM_read ((struct GNUNET_STREAM_Socket *) cls,
-                                        GNUNET_TIME_relative_multiply
-                                        (GNUNET_TIME_UNIT_SECONDS, 5),
-                                        &input_processor,
-                                        (void *) &peer2);
-  GNUNET_assert (NULL != peer2.io_handle);
-}
-
-
-/**
  * Functions of this type are called upon new stream connection from other peers
  *
- * @param cls the closure from GNUNET_STREAM_listen
+ * @param cls the PeerData of peer2
  * @param socket the socket representing the stream
  * @param initiator the identity of the peer who wants to establish a stream
  *            with us
@@ -285,35 +330,60 @@ stream_listen_cb (void *cls,
            struct GNUNET_STREAM_Socket *socket,
            const struct GNUNET_PeerIdentity *initiator)
 {
+  struct PeerData *peer=cls;
+  struct GNUNET_PeerIdentity self;
+
   GNUNET_assert (NULL != socket);
-  GNUNET_assert (NULL == initiator);   /* Local peer=NULL? */
   GNUNET_assert (socket != peer1.socket);
+  GNUNET_assert (&peer2 == peer);
+
+  /* Get our identity */
+  GNUNET_assert (GNUNET_OK == GNUNET_TESTING_get_peer_identity (config_peer1,
+                                                                &self));
+  GNUNET_assert (0 == memcmp (&self,
+                              initiator,
+                              sizeof (struct GNUNET_PeerIdentity)));
+  
+  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG,
+              "Peer connected: %s\n", GNUNET_i2s(initiator));
 
-  peer2.socket = socket;
-  read_task = GNUNET_SCHEDULER_add_now (&stream_read, (void *) socket);
+  peer->socket = socket;
+  peer->bytes_read = 0;
+  GNUNET_SCHEDULER_add_now (&stream_read_task, &peer2);
   return GNUNET_OK;
 }
 
 
 /**
  * Testing function
+ *
+ * @param cls NULL
+ * @param tc the task context
  */
 static void
 test (void *cls, const struct GNUNET_SCHEDULER_TaskContext *tc)
 {
+  struct GNUNET_PeerIdentity self;
+
   test_task = GNUNET_SCHEDULER_NO_TASK;
+  /* Get our identity */
+  GNUNET_assert (GNUNET_OK == GNUNET_TESTING_get_peer_identity (config_peer1,
+                                                                &self));
+
+  peer2_listen_socket = GNUNET_STREAM_listen (config_peer2,
+                                              10, /* App port */
+                                              &stream_listen_cb,
+                                              &peer2);
+  GNUNET_assert (NULL != peer2_listen_socket);
 
   /* Connect to stream library */
-  peer1.socket = GNUNET_STREAM_open (NULL,         /* Null for local peer? */
+  peer1.socket = GNUNET_STREAM_open (config_peer1,
+                                     &self,         /* Null for local peer? */
                                      10,           /* App port */
                                      &stream_open_cb,
-                                     (void *) &peer1);
+                                     &peer1,
+				     GNUNET_STREAM_OPTION_END);
   GNUNET_assert (NULL != peer1.socket);
-  peer2_listen_socket = GNUNET_STREAM_listen (10 /* App port */
-                                              &stream_listen_cb,
-                                              NULL);
-  GNUNET_assert (NULL != peer2_listen_socket);
-                  
 }
 
 /**
@@ -330,6 +400,9 @@ run (void *cls, char *const *args, const char *cfgfile,
                     "WARNING",
 #endif
                     NULL);
+   /* Duplicate the configuration */
+   config_peer1 = GNUNET_CONFIGURATION_dup (cfg);
+   config_peer2 = GNUNET_CONFIGURATION_dup (cfg);
    arm_pid =
      GNUNET_OS_start_process (GNUNET_YES, NULL, NULL, "gnunet-service-arm",
                               "gnunet-service-arm",
@@ -340,11 +413,10 @@ run (void *cls, char *const *args, const char *cfgfile,
 
    abort_task =
      GNUNET_SCHEDULER_add_delayed (GNUNET_TIME_relative_multiply
-                                   (GNUNET_TIME_UNIT_SECONDS, 20), &do_abort,
+                                   (GNUNET_TIME_UNIT_SECONDS, 60), &do_abort,
                                     NULL);
-
-   test_task = GNUNET_SCHEDULER_add_now (&test, (void *) cfg);
-
+   
+   test_task = GNUNET_SCHEDULER_add_now (&test, NULL);
 }
 
 /**
@@ -355,7 +427,7 @@ int main (int argc, char **argv)
   int ret;
 
   char *const argv2[] = { "test-stream-local",
-                          "-c", "test_stream.conf",
+                          "-c", "test_stream_local.conf",
 #if VERBOSE
                           "-L", "DEBUG",
 #endif
@@ -365,7 +437,7 @@ int main (int argc, char **argv)
   struct GNUNET_GETOPT_CommandLineOption options[] = {
     GNUNET_GETOPT_OPTION_END
   };
-  
+
   ret =
       GNUNET_PROGRAM_run ((sizeof (argv2) / sizeof (char *)) - 1, argv2,
                           "test-stream-local", "nohelp", options, &run, NULL);
@@ -381,6 +453,6 @@ int main (int argc, char **argv)
     GNUNET_log (GNUNET_ERROR_TYPE_WARNING, "test failed\n");
     return 1;
   }
-  GNUNET_log (GNUNET_ERROR_TYPE_DEBUG, "test ok\n");
+  GNUNET_log (GNUNET_ERROR_TYPE_INFO, "test ok\n");
   return 0;
 }
diff --git a/src/stream/test_stream_local.conf b/src/stream/test_stream_local.conf
index 3d955f0..884ecbc 100644
--- a/src/stream/test_stream_local.conf
+++ b/src/stream/test_stream_local.conf
@@ -9,7 +9,7 @@ DEBUG = YES
 AUTOSTART = YES
 ACCEPT_FROM = 127.0.0.1;
 HOSTNAME = localhost
-PORT = 10511
+PORT = 10700
 # PREFIX = valgrind --leak-check=full
 # PREFIX = xterm -geometry 100x85 -T peer1 -e gdb --args
 
@@ -19,7 +19,7 @@ AUTOSTART = YES
 ACCEPT_FROM6 = ::1;
 ACCEPT_FROM = 127.0.0.1;
 HOSTNAME = localhost
-PORT = 2100
+PORT = 12100
 
 [block]
 plugins = dht test
@@ -53,14 +53,20 @@ TIMEOUT = 300 s
 PORT = 12368
 
 [TESTING]
+NUM_PEERS = 5
 WEAKRANDOM = YES
+DEBUG = YES
+HOSTKEYSFILE = ../../contrib/testing_hostkeys.dat
+MAX_CONCURRENT_SSH = 10
+USE_PROGRESSBARS = YES
+PEERGROUP_TIMEOUT = 2400 s
 
 [gnunetd]
 HOSTKEY = $SERVICEHOME/.hostkey
 
 [PATHS]
-DEFAULTCONFIG = test_mesh.conf
-SERVICEHOME = /tmp/test-mesh/
+DEFAULTCONFIG = test_stream_local.conf
+SERVICEHOME = /tmp/test-stream/
 
 [dns]
 AUTOSTART = NO