Skip to content

Commit a6c43f2

Browse files
DavidBaduraDavidBadura
committed
refactor stream
1 parent 657473d commit a6c43f2

41 files changed

Lines changed: 567 additions & 1569 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎src/Console/Command/StoreMigrateCommand.php‎

Lines changed: 6 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,6 @@
66

77
use Patchlevel\EventSourcing\Console\InputHelper;
88
use Patchlevel\EventSourcing\Console\OutputStyle;
9-
use Patchlevel\EventSourcing\Message\Pipe;
109
use Patchlevel\EventSourcing\Message\Translator\Translator;
1110
use Patchlevel\EventSourcing\Store\Store;
1211
use Symfony\Component\Console\Attribute\AsCommand;
@@ -52,32 +51,17 @@ protected function execute(InputInterface $input, OutputInterface $output): int
5251
$style->info('Migration initialization...');
5352

5453
$count = $this->store->count();
55-
$messages = $this->store->load();
54+
$stream = $this->store->load();
5655

5756
$style->progressStart($count);
5857

59-
$bufferedMessages = [];
58+
$translatedStream = $stream->transform(...$this->translators);
6059

61-
$pipe = new Pipe(
62-
$messages,
63-
...$this->translators,
64-
);
60+
foreach ($translatedStream->chunk($buffer) as $chunk) {
61+
$messages = $chunk->toList();
6562

66-
foreach ($pipe as $message) {
67-
$bufferedMessages[] = $message;
68-
69-
if (count($bufferedMessages) < $buffer) {
70-
continue;
71-
}
72-
73-
$this->newStore->save(...$bufferedMessages);
74-
$bufferedMessages = [];
75-
$style->progressAdvance($buffer);
76-
}
77-
78-
if (count($bufferedMessages) !== 0) {
79-
$this->newStore->save(...$bufferedMessages);
80-
$style->progressAdvance(count($bufferedMessages));
63+
$this->newStore->save(...$messages);
64+
$style->progressAdvance(count($messages));
8165
}
8266

8367
$style->progressFinish();

‎src/Message/Pipe.php‎

Lines changed: 0 additions & 59 deletions
This file was deleted.

‎src/Message/Stream.php‎

Lines changed: 185 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,185 @@
1+
<?php
2+
3+
declare(strict_types=1);
4+
5+
namespace Patchlevel\EventSourcing\Message;
6+
7+
use ArrayIterator;
8+
use Generator;
9+
use Iterator;
10+
use Patchlevel\EventSourcing\Message\Translator\ChainTranslator;
11+
use Patchlevel\EventSourcing\Message\Translator\Translator;
12+
13+
use function iterator_to_array;
14+
15+
/** @implements Iterator<int, Message> */
16+
final class Stream implements Iterator
17+
{
18+
/** @var Iterator<int, Message>|null */
19+
private Iterator|null $iterator;
20+
21+
/** @var positive-int|0|null */
22+
private int|null $position = null;
23+
24+
private int|null $index = null;
25+
26+
/** @param iterable<int, Message> $messages */
27+
public function __construct(iterable $messages = [])
28+
{
29+
if ($messages instanceof Iterator) {
30+
$this->iterator = $messages;
31+
} else {
32+
$this->iterator = new ArrayIterator($messages);
33+
}
34+
}
35+
36+
public function close(): void
37+
{
38+
$this->iterator = null;
39+
$this->position = null;
40+
$this->index = null;
41+
}
42+
43+
public function current(): Message|null
44+
{
45+
$this->assertNotClosed();
46+
47+
return $this->iterator->valid() ? $this->iterator->current() : null;
48+
}
49+
50+
public function key(): int|null
51+
{
52+
$this->assertNotClosed();
53+
54+
if (!$this->iterator->valid()) {
55+
return null;
56+
}
57+
58+
return $this->iterator->key();
59+
}
60+
61+
public function next(): void
62+
{
63+
$this->assertNotClosed();
64+
$this->initialize();
65+
66+
$this->iterator->next();
67+
68+
if (!$this->iterator->valid()) {
69+
return;
70+
}
71+
72+
$this->position = $this->position === null ? 0 : $this->position + 1;
73+
$this->index = $this->iterator->key();
74+
}
75+
76+
public function rewind(): void
77+
{
78+
$this->assertNotClosed();
79+
$this->iterator->rewind();
80+
$this->position = null;
81+
$this->index = null;
82+
}
83+
84+
public function valid(): bool
85+
{
86+
$this->assertNotClosed();
87+
88+
return $this->iterator->valid();
89+
}
90+
91+
/** @phpstan-assert !null $this->iterator */
92+
private function assertNotClosed(): void
93+
{
94+
if ($this->iterator === null) {
95+
throw new StreamClosed();
96+
}
97+
}
98+
99+
public function end(): bool
100+
{
101+
$this->assertNotClosed();
102+
103+
return !$this->iterator->valid();
104+
}
105+
106+
public function position(): int|null
107+
{
108+
$this->assertNotClosed();
109+
$this->initialize();
110+
111+
return $this->position;
112+
}
113+
114+
public function index(): int|null
115+
{
116+
$this->assertNotClosed();
117+
$this->initialize();
118+
119+
return $this->index;
120+
}
121+
122+
private function initialize(): void
123+
{
124+
if (!$this->iterator->valid() || $this->position !== null) {
125+
return;
126+
}
127+
128+
$this->position = 0;
129+
$this->index = $this->iterator->key();
130+
}
131+
132+
/** @return list<Message> */
133+
public function toList(): array
134+
{
135+
return iterator_to_array($this, false);
136+
}
137+
138+
/** @return array<int, Message> */
139+
public function toArray(): array
140+
{
141+
return iterator_to_array($this);
142+
}
143+
144+
public function transform(Translator ...$translators): Stream
145+
{
146+
$chainTranslator = new ChainTranslator($translators);
147+
148+
$generator = function () use ($chainTranslator) {
149+
foreach ($this as $message) {
150+
foreach ($chainTranslator($message) as $translatedMessage) {
151+
yield $translatedMessage;
152+
}
153+
}
154+
};
155+
156+
return new Stream($generator());
157+
}
158+
159+
/** @return Generator<Stream> */
160+
public function chunk(int $size = 1000): Generator
161+
{
162+
$buffer = [];
163+
$bufferSize = 0;
164+
165+
foreach ($this as $message) {
166+
$buffer[] = $message;
167+
$bufferSize++;
168+
169+
if ($bufferSize !== $size) {
170+
continue;
171+
}
172+
173+
yield new Stream($buffer);
174+
175+
$buffer = [];
176+
$bufferSize = 0;
177+
}
178+
179+
if ($bufferSize <= 0) {
180+
return;
181+
}
182+
183+
yield new Stream($buffer);
184+
}
185+
}

src/Store/StreamClosed.php renamed to src/Message/StreamClosed.php

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,7 @@
22

33
declare(strict_types=1);
44

5-
namespace Patchlevel\EventSourcing\Store;
5+
namespace Patchlevel\EventSourcing\Message;
66

77
use RuntimeException;
88

‎src/Repository/DefaultRepository.php‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@
99
use Patchlevel\EventSourcing\EventBus\EventBus;
1010
use Patchlevel\EventSourcing\Identifier\Identifier;
1111
use Patchlevel\EventSourcing\Message\Message;
12+
use Patchlevel\EventSourcing\Message\Stream;
1213
use Patchlevel\EventSourcing\Metadata\AggregateRoot\AggregateRootMetadata;
1314
use Patchlevel\EventSourcing\Repository\MessageDecorator\MessageDecorator;
1415
use Patchlevel\EventSourcing\Snapshot\SnapshotNotFound;
@@ -23,7 +24,6 @@
2324
use Patchlevel\EventSourcing\Store\Header\RecordedOnHeader;
2425
use Patchlevel\EventSourcing\Store\Header\StreamNameHeader;
2526
use Patchlevel\EventSourcing\Store\Store;
26-
use Patchlevel\EventSourcing\Store\Stream;
2727
use Patchlevel\EventSourcing\Store\StreamStartHeader;
2828
use Patchlevel\EventSourcing\Store\UniqueConstraintViolation;
2929
use Psr\Clock\ClockInterface;

‎src/Store/AppendStore.php‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
namespace Patchlevel\EventSourcing\Store;
66

77
use Patchlevel\EventSourcing\Message\Message;
8+
use Patchlevel\EventSourcing\Message\Stream;
89

910
/** @experimental */
1011
interface AppendStore

0 commit comments

Comments
 (0)