Fix #54 : Added new stream events that indicate end of reading/writing

src/Stream/AsyncStream.php

@@ -188,7 +188,7 @@ public function close()
     public function handleWrite()
     {
         $text = $this->buffer->peek();
-        $sent = fwrite($this->resource, $text);
+        $sent = fwrite($this->resource, $text, $this->bufferSize);
 
         if ($sent === false)
         {
@@ -209,6 +209,7 @@ public function handleWrite()
             $this->loop->removeWriteStream($this->resource);
             $this->listening = false;
             $this->emit('drain', [ $this ]);
+            $this->emit('finish', [ $this ]);
         }
     }
 

src/Stream/AsyncStreamWriter.php

@@ -181,7 +181,7 @@ public function close()
     public function handleWrite()
     {
         $text = $this->buffer->peek();
-        $sent = fwrite($this->resource, $text);
+        $sent = fwrite($this->resource, $text, $this->bufferSize);
 
         if ($sent === false)
         {
@@ -202,6 +202,7 @@ public function handleWrite()
             $this->loop->removeWriteStream($this->resource);
             $this->listening = false;
             $this->emit('drain', [ $this ]);
+            $this->emit('finish', [ $this ]);
         }
 
         if ($this->closing)

src/Stream/Stream.php

@@ -108,6 +108,7 @@ public function write($text)
         }
 
         $this->emit('drain', [ $this ]);
+        $this->emit('finish', [ $this ]);
 
         return true;
     }
@@ -141,6 +142,11 @@ public function read($length = null)
         else if ($ret !== '')
         {
             $this->emit('data', [ $this, $ret ]);
+
+            if (strlen($ret) < $length)
+            {
+                $this->emit('end', [ $this ]);
+            }
         }
 
         return $ret;

src/Stream/StreamReader.php

@@ -97,6 +97,11 @@ public function read($length = null)
         else if ($ret !== '')
         {
             $this->emit('data', [ $this, $ret ]);
+
+            if (strlen($ret) < $length)
+            {
+                $this->emit('end', [ $this ]);
+            }
         }
 
         return $ret;

src/Stream/StreamReaderInterface.php

@@ -6,6 +6,7 @@
 
 /**
  * @event data : callable(object, string)
+ * @event end  : callable(object)
  */
 interface StreamReaderInterface extends EventEmitterInterface, StreamSeekerInterface
 {

src/Stream/StreamWriter.php

@@ -91,6 +91,7 @@ public function write($text)
         }
 
         $this->emit('drain', [ $this ]);
+        $this->emit('finish', [ $this ]);
 
         return true;
     }

src/Stream/StreamWriterInterface.php

@@ -5,7 +5,8 @@
 use Kraken\Event\EventEmitterInterface;
 
 /**
- * @event drain : callable(object)
+ * @event drain  : callable(object)
+ * @event finish : callable(object)
  */
 interface StreamWriterInterface extends EventEmitterInterface, StreamSeekerInterface
 {

test/_Module/Stream/AsyncStreamTest.php

@@ -41,7 +41,9 @@ public function testAsyncStream_WritesAndReadsDataCorrectly($readerClass, $write
                     $sim->expect('data', $data);
                     $conn->close();
                 });
+                $reader->on('end', $this->expectCallableOnce());
                 $reader->on('drain', $this->expectCallableNever());
+                $reader->on('finish', $this->expectCallableNever());
                 $reader->on('error', $this->expectCallableNever());
                 $reader->on('close', function() use($sim, &$cnt) {
                     $sim->expect('close');
@@ -56,10 +58,12 @@ public function testAsyncStream_WritesAndReadsDataCorrectly($readerClass, $write
                     $loop
                 );
                 $writer->on('data', $this->expectCallableNever());
+                $writer->on('end', $this->expectCallableNever());
                 $writer->on('drain', function(AsyncStreamWriterInterface $writer) use($sim) {
                     $sim->expect('drain');
                     $writer->close();
                 });
+                $writer->on('finish', $this->expectCallableOnce());
                 $writer->on('error', $this->expectCallableNever());
                 $writer->on('close', function() use($sim, &$cnt) {
                     $sim->expect('close');

test/_Module/Stream/StreamTest.php

@@ -34,10 +34,12 @@ public function testStream_WriteAndReadDataScenario()
         $reader->on('data', function($origin, $data) use(&$capturedData) {
             $capturedData = $data;
         });
+        $reader->on('end', $this->expectCallableOnce());
         $reader->on('error', $this->expectCallableNever());
         $reader->on('close', $this->expectCallableOnce());
 
         $writer->on('drain', $this->expectCallableOnce());
+        $writer->on('finish', $this->expectCallableOnce());
         $writer->on('error', $this->expectCallableNever());
         $writer->on('close', $this->expectCallableOnce());
 

test/_Unit/Stream/AsyncStreamReaderTest.php

@@ -2,14 +2,17 @@
 
 namespace Kraken\_Unit\Stream;
 
+use Kraken\Loop\Loop;
 use Kraken\Loop\LoopInterface;
+use Kraken\Loop\Model\SelectLoop;
 use Kraken\Stream\AsyncStreamReader;
 
 class AsyncStreamReaderTest extends StreamSeekerTest
 {
     public function testApiRead_ReadsDataProperly()
     {
-        $stream = $this->createAsyncStreamReaderMock();
+        $loop = new Loop(new SelectLoop);
+        $stream = $this->createAsyncStreamReaderMock(null, $loop);
         $resource = $stream->getResource();
 
         $expectedData = "foobar\n";
@@ -20,10 +23,15 @@ public function testApiRead_ReadsDataProperly()
             $capturedOrigin = $origin;
             $capturedData = $data;
         });
+        $stream->on('end', $this->expectCallableOnce());
 
         fwrite($resource, $expectedData);
         rewind($resource);
-        $stream->handleData($stream->getResource());
+
+        $loop->addTimer(1e-1, function() use($loop) {
+            $loop->stop();
+        });
+        $loop->start();
 
         $this->assertSame($expectedData, $capturedData);
         $this->assertSame($stream, $capturedOrigin);

test/_Unit/Stream/AsyncStreamWriterTest.php

@@ -2,24 +2,34 @@
 
 namespace Kraken\_Unit\Stream;
 
+use Kraken\Loop\Model\SelectLoop;
+use Kraken\Loop\Loop;
 use Kraken\Loop\LoopInterface;
 use Kraken\Stream\AsyncStreamWriter;
 
 class AsyncStreamWriterTest extends StreamSeekerTest
 {
     public function testApiWrite_WritesDataProperly()
     {
-        $stream = $this->createAsyncStreamWriterMock();
+        $loop = new Loop(new SelectLoop);
+        $stream = $this->createAsyncStreamWriterMock(null, $loop);
         $resource = $stream->getResource();
 
-        $expectedData = "foobar\n";
+        $expectedData = str_repeat('X', (int) $stream->getBufferSize()*1.5);
 
-        $stream->on('drain', $this->expectCallableOnce());
+        $stream->on('drain', $this->expectCallableTwice());
+        $stream->on('finish', $this->expectCallableOnce());
 
-        $stream->write($expectedData);
-        $stream->rewind();
+        $stream->write(substr($expectedData, 0, 1024));
+        $stream->write(substr($expectedData, 1024));
+
+        $loop->addTimer(1e-1, function() use($loop) {
+            $loop->stop();
+        });
+        $loop->start();
 
-        $this->assertSame($expectedData, fread($resource, $stream->getBufferSize()));
+        $stream->rewind();
+        $this->assertSame($expectedData, fread($resource, (int) $stream->getBufferSize()*2));
     }
 
     /**

test/_Unit/Stream/StreamReaderTest.php

@@ -45,6 +45,7 @@ public function testApiRead_ReadsDataCorrectly()
             $capturedOrigin = $origin;
             $capturedData = $data;
         });
+        $stream->on('end', $this->expectCallableOnce());
 
         fwrite($resource, $expectedData);
         rewind($resource);

test/_Unit/Stream/StreamTest.php

@@ -32,6 +32,7 @@ public function testApiRead_ReadsDataCorrectly()
             $capturedOrigin = $origin;
             $capturedData = $data;
         });
+        $stream->on('end', $this->expectCallableOnce());
 
         fwrite($resource, $expectedData);
         rewind($resource);

test/_Unit/Stream/StreamWriterTest.php

@@ -40,9 +40,11 @@ public function testApiWrite_WritesDataCorrectly()
         $expectedData = "foobar\n";
         $capturedData = null;
 
-        $stream->on('drain', $this->expectCallableOnce());
+        $stream->on('drain', $this->expectCallableTwice());
+        $stream->on('finish', $this->expectCallableTwice());
 
-        $stream->write($expectedData);
+        $stream->write(substr($expectedData, 0, 2));
+        $stream->write(substr($expectedData, 2));
         $stream->rewind();
 
         $this->assertSame($expectedData, fread($resource, $stream->getBufferSize()));