Merge pull request #181 from php-enqueue/fs-edge-case-with-message-size

[fs] fix bug that happens with specific message length.

Author: makasim

pkg/fs/FsConsumer.php

@@ -164,7 +164,6 @@ private function readFrame($file, $frameNumber)
 
         fseek($file, -$offset, SEEK_END);
         $frame = fread($file, $frameSize);
-
         if ('' == $frame) {
             return '';
         }
@@ -173,6 +172,12 @@ private function readFrame($file, $frameNumber)
             return $frame;
         }
 
-        return $this->readFrame($file, $frameNumber + 1).$frame;
+        $previousFrame = $this->readFrame($file, $frameNumber + 1);
+
+        if ('|' === substr($previousFrame, -1) && '{' === $frame[0]) {
+            return '|'.$frame;
+        }
+
+        return $previousFrame.$frame;
     }
 }

pkg/fs/Tests/Functional/FsConsumerTest.php

@@ -5,7 +5,6 @@
 use Enqueue\Fs\FsConnectionFactory;
 use Enqueue\Fs\FsContext;
 use Enqueue\Fs\FsMessage;
-use Makasim\File\TempFile;
 use PHPUnit\Framework\TestCase;
 
 class FsConsumerTest extends TestCase
@@ -19,8 +18,7 @@ public function setUp()
     {
         $this->fsContext = (new FsConnectionFactory(['path' => sys_get_temp_dir()]))->createContext();
 
-        new TempFile(sys_get_temp_dir().'/fs_test_queue');
-        file_put_contents(sys_get_temp_dir().'/fs_test_queue', '');
+        $this->fsContext->purge($this->fsContext->createQueue('fs_test_queue'));
     }
 
     public function tearDown()
@@ -68,4 +66,29 @@ public function testShouldConsumeMessagesFromFileOneByOne()
 
         $this->assertEmpty(file_get_contents(sys_get_temp_dir().'/fs_test_queue'));
     }
+
+    /**
+     * @group bug
+     */
+    public function testShouldNotFailOnSpecificMessageSize()
+    {
+        $context = $this->fsContext;
+        $queue = $context->createQueue('fs_test_queue');
+        $context->purge($queue);
+
+        $consumer = $context->createConsumer($queue);
+        $producer = $context->createProducer();
+
+        $producer->send($queue, $context->createMessage(str_repeat('a', 23)));
+        $producer->send($queue, $context->createMessage(str_repeat('b', 24)));
+
+        $message = $consumer->receiveNoWait();
+        $this->assertSame(str_repeat('b', 24), $message->getBody());
+
+        $message = $consumer->receiveNoWait();
+        $this->assertSame(str_repeat('a', 23), $message->getBody());
+
+        $message = $consumer->receiveNoWait();
+        $this->assertNull($message);
+    }
 }