When reading data, don't peek.

composer.json

@@ -31,7 +31,10 @@
     "psr-4": {"FreeDSx\\Socket\\": "src/FreeDSx/Socket"}
   },
   "autoload-dev": {
-    "psr-4": {"fixture\\FreeDSx\\Socket\\": "tests/fixture/FreeDSx/Socket"}
+    "psr-4": {
+      "fixture\\FreeDSx\\Socket\\": "tests/fixture/FreeDSx/Socket",
+      "spec\\FreeDSx\\Socket\\": "tests/spec/FreeDSx/Socket"
+    }
   },
   "scripts": {
     "test-coverage": [

src/FreeDSx/Socket/Queue/MessageQueue.php

@@ -13,6 +13,9 @@
 use FreeDSx\Socket\Exception\ConnectionException;
 use FreeDSx\Socket\Exception\PartialMessageException;
 use FreeDSx\Socket\Socket;
+use Generator;
+use function strlen;
+use function substr;
 
 /**
  * Used to retrieve Messages/PDUs from a socket.
@@ -46,43 +49,42 @@ public function __construct(Socket $socket)
 
     /**
      * @param int|null $id
-     * @return \Generator
+     * @return Generator
      * @throws ConnectionException
      */
-    public function getMessages(?int $id = null)
+    public function getMessages(?int $id = null): Generator
     {
-        if (!$this->hasBuffer()) {
-            $this->addToAvailableBufferOrFail();
+        while (true) {
+            yield $this->getMessage($id);
         }
+    }
 
-        while ($this->hasBuffer()) {
-            try {
-                if ($this->hasAvailableBuffer()) {
-                    $this->addToConsumableBuffer();
-                } elseif (!$this->hasConsumableBuffer()) {
-                    $this->addToAvailableBufferOrFail();
-                }
-            } catch (PartialMessageException $exception) {
+    /**
+     * @throws ConnectionException
+     * @throws PartialMessageException
+     */
+    private function readOneMessage(): Message
+    {
+        while (true) {
+            if (!$this->hasConsumableBuffer() && !$this->hasAvailableBuffer()) {
                 $this->addToAvailableBufferOrFail();
             }
 
+            if ($this->hasAvailableBuffer()) {
+                $this->addToConsumableBuffer();
+            }
+
             try {
-                while ($this->hasConsumableBuffer()) {
-                    $message = $this->consume();
-                    if ($message !== null) {
-                        yield $this->constructMessage($message, $id);
-                    }
-                }
-            } catch (PartialMessageException $e) {
-                if ($this->hasAvailableBuffer()) {
-                    $this->addToConsumableBuffer();
-                } else {
-                    $this->addToAvailableBufferOrFail();
-                }
+                return $this->consume();
+            } catch (PartialMessageException $exception) {
+                $this->addToAvailableBufferOrFail();
             }
         }
     }
 
+    /**
+     * @throws ConnectionException
+     */
     protected function addToAvailableBufferOrFail(): void
     {
         $bytes = $this->socket->read();
@@ -98,7 +100,7 @@ protected function addToConsumableBuffer(): void
     {
         if ($this->hasAvailableBuffer()) {
             $buffer = $this->unwrap((string)$this->buffer);
-            $this->buffer = \substr((string)$this->buffer, $buffer->endsAt());
+            $this->buffer = substr((string)$this->buffer, $buffer->endsAt());
             $this->toConsume .= $buffer->bytes();
         }
     }
@@ -110,12 +112,12 @@ protected function hasBuffer(): bool
 
     protected function hasAvailableBuffer(): bool
     {
-        return \strlen((string)$this->buffer) !== 0;
+        return strlen((string)$this->buffer) !== 0;
     }
 
     protected function hasConsumableBuffer(): bool
     {
-        return \strlen((string)$this->toConsume) !== 0;
+        return strlen((string)$this->toConsume) !== 0;
     }
 
     /**
@@ -129,11 +131,10 @@ protected function consume(): ?Message
         try {
             $message = $this->decode($this->toConsume);
             $lastPos = (int)$message->getLastPosition();
-            $this->toConsume = \substr($this->toConsume, $lastPos);
-
-            if ($this->toConsume === '' && ($peek = $this->socket->read(false)) !== false) {
-                $this->buffer .= $peek;
-            }
+            $this->toConsume = substr(
+                $this->toConsume,
+                $lastPos
+            );
         } catch (PartialMessageException $exception) {
             # If we have available buffer, it might have what we need. Attempt to add it. Otherwise let it bubble...
             if ($this->hasAvailableBuffer()) {
@@ -157,7 +158,10 @@ protected function consume(): ?Message
      */
     protected function unwrap($bytes) : Buffer
     {
-        return new Buffer($bytes, \strlen($bytes));
+        return new Buffer(
+            $bytes,
+            strlen($bytes)
+        );
     }
 
     /**
@@ -177,7 +181,10 @@ protected abstract function decode($bytes) : Message;
      */
     public function getMessage(?int $id = null)
     {
-        return $this->getMessages($id)->current();
+        return $this->constructMessage(
+            $this->readOneMessage(),
+            $id
+        );
     }
 
     /**
@@ -187,8 +194,10 @@ public function getMessage(?int $id = null)
      * @param int|null $id
      * @return mixed
      */
-    protected function constructMessage(Message $message, ?int $id = null)
-    {
+    protected function constructMessage(
+        Message $message,
+        ?int $id = null
+    ) {
         return $message->getMessage();
     }
 }

src/FreeDSx/Socket/Socket.php

@@ -11,6 +11,13 @@
 namespace FreeDSx\Socket;
 
 use FreeDSx\Socket\Exception\ConnectionException;
+use function array_merge;
+use function fread;
+use function in_array;
+use function stream_set_blocking;
+use function stream_set_timeout;
+use function stream_socket_client;
+use function stream_socket_enable_crypto;
 
 /**
  * Represents a generic socket.
@@ -103,8 +110,8 @@ class Socket
     public function __construct($resource = null, array $options = [])
     {
         $this->socket = $resource;
-        $this->options = \array_merge($this->options, $options);
-        if (!\in_array($this->options['transport'], self::TRANSPORTS, true)) {
+        $this->options = array_merge($this->options, $options);
+        if (!in_array($this->options['transport'], self::TRANSPORTS, true)) {
             throw new \RuntimeException(sprintf(
                 'The transport "%s" is not valid. It must be one of: %s',
                 $this->options['transport'],
@@ -122,19 +129,22 @@ public function __construct($resource = null, array $options = [])
      */
     public function read(bool $block = true)
     {
-        $data = false;
-
-        \stream_set_blocking($this->socket, $block);
-        while (\strlen((string) ($buffer = \fread($this->socket, $this->options['buffer_size']))) > 0) {
-            $data .= $buffer;
-            if ($block) {
-                $block = false;
-                \stream_set_blocking($this->socket, false);
-            }
+        stream_set_blocking($this->socket, $block);
+        $data = fread(
+            $this->socket,
+            $this->options['buffer_size']
+        );
+
+        if (!$block) {
+            stream_set_blocking(
+                $this->socket,
+                true
+            );
         }
-        \stream_set_blocking($this->socket, true);
 
-        return $data;
+        return $data === ''
+            ? false
+            : $data;
     }
 
     /**
@@ -154,7 +164,7 @@ public function write(string $data)
      */
     public function block(bool $block)
     {
-        \stream_set_blocking($this->socket, $block);
+        stream_set_blocking($this->socket, $block);
 
         return $this;
     }
@@ -210,9 +220,9 @@ public function close()
      */
     public function encrypt(bool $encrypt)
     {
-        \stream_set_blocking($this->socket, true);
-        $result = \stream_socket_enable_crypto($this->socket, $encrypt, $this->options['ssl_crypto_method']);
-        \stream_set_blocking($this->socket, false);
+        stream_set_blocking($this->socket, true);
+        $result = stream_socket_enable_crypto($this->socket, $encrypt, $this->options['ssl_crypto_method']);
+        stream_set_blocking($this->socket, false);
 
         if ((bool) $result == false) {
             throw new ConnectionException(sprintf(
@@ -244,7 +254,7 @@ public function connect(string $host)
             $uri .= ':' . $this->options['port'];
         }
 
-        $socket = @\stream_socket_client(
+        $socket = @stream_socket_client(
             $uri,
             $this->errorNumber,
             $this->errorMessage,
@@ -303,7 +313,7 @@ public static function unix(
     ): Socket {
         return self::create(
             $file,
-            \array_merge(
+            array_merge(
                 $options,
                 ['transport' => 'unix']
             )
@@ -320,7 +330,7 @@ public static function unix(
      */
     public static function tcp(string $host, array $options = []) : Socket
     {
-        return self::create($host, \array_merge($options, ['transport' => 'tcp']));
+        return self::create($host, array_merge($options, ['transport' => 'tcp']));
     }
 
     /**
@@ -333,7 +343,7 @@ public static function tcp(string $host, array $options = []) : Socket
      */
     public static function udp(string $host, array $options = []) : Socket
     {
-        return self::create($host, \array_merge($options, [
+        return self::create($host, array_merge($options, [
             'transport' => 'udp',
             'buffer_size' => 65507,
         ]));
@@ -351,7 +361,7 @@ protected function createSocketContext()
             }
         }
         if ($this->options['ssl_validate_cert'] === false) {
-            $sslOpts = \array_merge($sslOpts, [
+            $sslOpts = array_merge($sslOpts, [
                 'allow_self_signed' => true,
                 'verify_peer' => false,
                 'verify_peer_name' => false,
@@ -369,6 +379,6 @@ protected function createSocketContext()
      */
     protected function setStreamOpts() : void
     {
-        \stream_set_timeout($this->socket, $this->options['timeout_read']);
+        stream_set_timeout($this->socket, $this->options['timeout_read']);
     }
 }

tests/spec/FreeDSx/Socket/Queue/Asn1MessageQueueSpec.php

@@ -58,4 +58,45 @@ function it_should_throw_an_exception_on_get_message_when_there_is_none($socket)
 
         $this->shouldThrow(ConnectionException::class)->duringGetMessage();
     }
+
+    function it_should_not_peek_the_socket_after_decoding_a_complete_message($socket, $encoder)
+    {
+        $socket->read()->willReturn('foo');
+        $socket->read(false)->shouldNotBeCalled();
+        $encoder->decode('foo')->shouldBeCalled()->willReturn(new IntegerType(100));
+        $encoder->getLastPosition()->willReturn(3);
+
+        $this->getMessage()->shouldBeLike(new Pdu(new IntegerType(100)));
+    }
+
+    function it_should_yield_messages_continuously_from_the_generator($socket, $encoder)
+    {
+        $socket->read()->willReturn('foobar', 'baz');
+        $encoder->decode('foobar')->willReturn(new IntegerType(1));
+        $encoder->decode('bar')->willReturn(new IntegerType(2));
+        $encoder->decode('baz')->willReturn(new IntegerType(3));
+        $encoder->getLastPosition()->willReturn(3);
+
+        $iter = $this->getMessages()->getWrappedObject();
+
+        if (!$iter instanceof \Generator) {
+            throw new \RuntimeException('getMessages() must return a Generator.');
+        }
+
+        $first = $iter->current();
+        $iter->next();
+        $second = $iter->current();
+        $iter->next();
+        $third = $iter->current();
+
+        if ($first != new Pdu(new IntegerType(1))) {
+            throw new \RuntimeException('First yielded message did not match.');
+        }
+        if ($second != new Pdu(new IntegerType(2))) {
+            throw new \RuntimeException('Second yielded message did not match.');
+        }
+        if ($third != new Pdu(new IntegerType(3))) {
+            throw new \RuntimeException('Third yielded message did not match.');
+        }
+    }
 }

tests/spec/FreeDSx/Socket/RequiresUnixTransport.php

@@ -0,0 +1,24 @@
+<?php
+
+/**
+ * This file is part of the FreeDSx Socket package.
+ *
+ * (c) Chad Sikorra <Chad.Sikorra@gmail.com>
+ *
+ * For the full copyright and license information, please view the LICENSE
+ * file that was distributed with this source code.
+ */
+
+namespace spec\FreeDSx\Socket;
+
+use PhpSpec\Exception\Example\SkippingException;
+
+trait RequiresUnixTransport
+{
+    private function requireUnixTransport(): void
+    {
+        if (!in_array('unix', stream_get_transports(), true)) {
+            throw new SkippingException('The "unix" stream transport is not available on this platform.');
+        }
+    }
+}