Update Noop example for ChanReader

CREATING_TOXICS.md

@@ -145,19 +145,13 @@ An implementation of the noop toxic above using the stream package would look so
 ```go
 func (t *NoopToxic) Pipe(stub *toxics.ToxicStub) {
     buf := make([]byte, 32*1024)
-    writer := stream.NewChanWriter(stub.Output)
-    reader := stream.NewChanReader(stub.Input)
-    reader.SetInterrupt(stub.Interrupt)
     for {
-        n, err := reader.Read(buf)
-        if err == stream.ErrInterrupted {
-            writer.Write(buf[:n])
-            return
-        } else if err == io.EOF {
-            stub.Close()
+        n, err := stub.Reader.Read(buf)
+        if stub.HandleReadError(err) {
             return
         }
-        writer.Write(buf[:n])
+        stub.Writer.Write(buf[:n])
+        stub.Reader.Checkpoint(0)
     }
 }
 ```

stream/io_chan_test.go

@@ -380,6 +380,20 @@ func TestReadWriterFlush(t *testing.T) {
 	AssertClosed(t, rw, []byte("rld"))
 }
 
+func TestReadWriterDoubleFlush(t *testing.T) {
+	rw := NewTestReadWriter()
+	buf := make([]byte, 8)
+
+	AssertRead(t, rw, buf, "hello wo", nil)
+	rw.FlushTo(rw)
+	rw.FlushTo(rw)
+	AssertRead(t, rw, buf, "foobar", nil)
+	rw.FlushTo(rw)
+	AssertRead(t, rw, buf, "", io.EOF)
+
+	AssertClosed(t, rw, []byte("rld"))
+}
+
 func TestReadWriterNoCheckpoint(t *testing.T) {
 	rw := NewTestReadWriter()
 	buf := make([]byte, 32)