{"action":"opened","issue":{"url":"https://api.github.com/repos/Rxnet/eventstore-client/issues/16","repository_url":"https://api.github.com/repos/Rxnet/eventstore-client","labels_url":"https://api.github.com/repos/Rxnet/eventstore-client/issues/16/labels{/name}","comments_url":"https://api.github.com/repos/Rxnet/eventstore-client/issues/16/comments","events_url":"https://api.github.com/repos/Rxnet/eventstore-client/issues/16/events","html_url":"https://github.com/Rxnet/eventstore-client/issues/16","id":342332133,"node_id":"MDU6SXNzdWUzNDIzMzIxMzM=","number":16,"title":"Fails to read last page","user":{"login":"callistino","id":3854713,"node_id":"MDQ6VXNlcjM4NTQ3MTM=","avatar_url":"https://avatars0.githubusercontent.com/u/3854713?v=4","gravatar_id":"","url":"https://api.github.com/users/callistino","html_url":"https://github.com/callistino","followers_url":"https://api.github.com/users/callistino/followers","following_url":"https://api.github.com/users/callistino/following{/other_user}","gists_url":"https://api.github.com/users/callistino/gists{/gist_id}","starred_url":"https://api.github.com/users/callistino/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/callistino/subscriptions","organizations_url":"https://api.github.com/users/callistino/orgs","repos_url":"https://api.github.com/users/callistino/repos","events_url":"https://api.github.com/users/callistino/events{/privacy}","received_events_url":"https://api.github.com/users/callistino/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":0,"created_at":"2018-07-18T13:37:15Z","updated_at":"2018-07-18T13:37:15Z","closed_at":null,"author_association":"NONE","body":"Reading from the store using `\\Rxnet\\EventStore\\EventStore::readEventsForward` fails to read the last page of events. I've narrowed it down to this method `\\Rxnet\\EventStore\\EventStore::readEvents`. Specifically the observable's `takeWhile`. This method exits as soon as the stream reaches the end or if it has read the total count of events  requested and exists without passing forward the current set of events if any. This happens even if the last page of event is the first page. So if you have less than 10 events it will not read any.\r\n\r\nI've prepared an example:\r\n\r\n1. Run `php examples/write.php`. Take note of the last event written or connect to the web ui at [localhost:2113](localhost:2113).\r\n2. Copy the snippet below to *examples/readForward.php*.\r\n3. Run `php examples/readForward.php`\r\n4. See the last \"received\" event is either 10 less than the total available or none if the total is less than 10.\r\n\r\n```php\r\n<?php\r\n\r\nuse EventLoop\\EventLoop;\r\n\r\nrequire __DIR__.'/../vendor/autoload.php';\r\n\r\n\r\n$eventStore = new \\Rxnet\\EventStore\\EventStore();\r\n$eventStore->connect()\r\n    ->subscribe(function() use ($eventStore) {\r\n        echo \"connected \\n\";\r\n        $eventStore->readEventsForward('domain-test.fr')\r\n            ->subscribe(function (\\Rxnet\\EventStore\\EventRecord $record) {\r\n                echo \"received {$record->getId()}  {$record->getNumber()}@{$record->getStreamId()} {$record->getType()} created at {$record->getCreated()->format('c')}\\n\";\r\n            });\r\n    }, function (\\Exception $e) {\r\n        echo $e->getMessage();\r\n    });\r\n\r\nEventLoop::getLoop()->run();\r\n```\r\n\r\nBelow is a very crude possible fix I've implemented. I've researched and I haven't found a better way to use observable operators instead of using flags. \r\n\r\n```bash\r\ndiff --git a/src/Rxnet/EventStore/EventStore.php b/src/Rxnet/EventStore/EventStore.php\r\nindex 48f0a89..4243af5 100644\r\n--- a/src/Rxnet/EventStore/EventStore.php\r\n+++ b/src/Rxnet/EventStore/EventStore.php\r\n@@ -538,6 +538,7 @@ class EventStore\r\n         // OnDemand ? onBackpressureBuffer ?\r\n         return Observable::create(function (ObserverInterface $observer) use ($messageType, $query) {\r\n             $end = false;\r\n+            $stop = false;\r\n             $maxPossible = 10; //4096\r\n             $max = ($query instanceof ReadStreamEvents) ? $query->getMaxCount() : self::POSITION_LATEST;\r\n\r\n@@ -559,7 +560,8 @@ class EventStore\r\n                     return Observable::of($event);\r\n                 })\r\n                 // If more data is needed do another query\r\n-                ->doOnNext(function (ReadStreamEventsCompleted $event) use ($query, $correlationID, &$end, &$asked, $max, $maxPossible, $messageType) {\r\n+                ->doOnNext(function (ReadStreamEventsCompleted $event) use ($query, $correlationID, &$end, &$stop, &$asked, $max, $maxPossible, $messageType) {\r\n+                   $stope = $end;\r\n                     $records = $event->getEvents();\r\n                     $asked -= count($records);\r\n                     if ($event->getIsEndOfStream()) {\r\n@@ -567,7 +569,7 @@ class EventStore\r\n                     } elseif ($asked <= 0 && $max != self::POSITION_LATEST) {\r\n                         $end = true;\r\n                     }\r\n-                    if (!$end) {\r\n+                    if (!$stop && count($records) > 0) {\r\n                         $start = $records[count($records) - 1];\r\n                         /* @var ResolvedIndexedEvent $start */\r\n\r\n@@ -589,8 +591,8 @@ class EventStore\r\n                     }\r\n                 })\r\n                 // Continue to watch until we have all our results (or end)\r\n-                ->takeWhile(function () use (&$end) {\r\n-                    return !$end;\r\n+                ->takeWhile(function () use (&$stop) {\r\n+                    return !$stop;\r\n                 })\r\n                 // Format EventRecord for easy reading\r\n                 ->flatMap(function (ReadStreamEventsCompleted $event) use (&$asked, &$end) {\r\n@@ -608,4 +610,4 @@ class EventStore\r\n                 ->subscribe($observer);\r\n         });\r\n     }\r\n-}\r\n\\ No newline at end of file\r\n+}\r\n```"}}
{"actor":{"display_login":"callistino"}}