@@ 83-107 (lines=25) @@ | ||
80 | * @param string $streamId |
|
81 | * @return EventStreamInterface |
|
82 | */ |
|
83 | public function readFullStream($streamId) |
|
84 | { |
|
85 | $stmt = $this->connection->prepare( |
|
86 | 'SELECT * |
|
87 | FROM events |
|
88 | WHERE stream_id = :streamId' |
|
89 | ); |
|
90 | $stmt->bindValue(':streamId', $streamId); |
|
91 | $stmt->execute(); |
|
92 | $results = $stmt->fetchAll(); |
|
93 | ||
94 | $storedEvents = array_map(function($result) { |
|
95 | return new StoredEvent( |
|
96 | $result['id'], |
|
97 | $result['stream_id'], |
|
98 | $result['type'], |
|
99 | $result['event'], |
|
100 | $result['metadata'], |
|
101 | new \DateTimeImmutable($result['occurred_on']), |
|
102 | Version::fromString($result['version']) |
|
103 | ); |
|
104 | }, $results); |
|
105 | ||
106 | return $this->domainEventStreamFromStoredEvents($storedEvents); |
|
107 | } |
|
108 | ||
109 | /** |
|
110 | * @return EventStreamInterface[] |
|
@@ 130-152 (lines=23) @@ | ||
127 | /** |
|
128 | * @return EventStreamInterface |
|
129 | */ |
|
130 | public function readAllEvents() |
|
131 | { |
|
132 | $stmt = $this->connection->prepare( |
|
133 | 'SELECT * |
|
134 | FROM events' |
|
135 | ); |
|
136 | $stmt->execute(); |
|
137 | $results = $stmt->fetchAll(); |
|
138 | ||
139 | $storedEvents = array_map(function($result) { |
|
140 | return new StoredEvent( |
|
141 | $result['id'], |
|
142 | $result['stream_id'], |
|
143 | $result['type'], |
|
144 | $result['event'], |
|
145 | $result['metadata'], |
|
146 | new \DateTimeImmutable($result['occurred_on']), |
|
147 | Version::fromString($result['version']) |
|
148 | ); |
|
149 | }, $results); |
|
150 | ||
151 | return $this->domainEventStreamFromStoredEvents($storedEvents); |
|
152 | } |
|
153 | ||
154 | /** |
|
155 | * @param string $type |
|
@@ 159-185 (lines=27) @@ | ||
156 | * @param Version $version |
|
157 | * @return EventStreamInterface |
|
158 | */ |
|
159 | protected function readStoredEventsOfTypeAndVersion($type, $version) |
|
160 | { |
|
161 | $stmt = $this->connection->prepare( |
|
162 | 'SELECT * |
|
163 | FROM events |
|
164 | WHERE type = :type |
|
165 | AND version = :version' |
|
166 | ); |
|
167 | $stmt->bindValue(':type', $type); |
|
168 | $stmt->bindValue(':version', $version); |
|
169 | $stmt->execute(); |
|
170 | $results = $stmt->fetchAll(); |
|
171 | ||
172 | $storedEvents = array_map(function($result) { |
|
173 | return new StoredEvent( |
|
174 | $result['id'], |
|
175 | $result['stream_id'], |
|
176 | $result['type'], |
|
177 | $result['event'], |
|
178 | $result['metadata'], |
|
179 | new \DateTimeImmutable($result['occurred_on']), |
|
180 | Version::fromString($result['version']) |
|
181 | ); |
|
182 | }, $results); |
|
183 | ||
184 | return new EventStream($storedEvents); |
|
185 | } |
|
186 | ||
187 | /** |
|
188 | * @param string $streamId |
@@ 84-108 (lines=25) @@ | ||
81 | * @param string $streamId |
|
82 | * @return EventStreamInterface |
|
83 | */ |
|
84 | public function readFullStream($streamId) |
|
85 | { |
|
86 | $stmt = $this->connection->prepare( |
|
87 | 'SELECT * |
|
88 | FROM events |
|
89 | WHERE stream_id = :streamId' |
|
90 | ); |
|
91 | $stmt->bindValue(':streamId', $streamId); |
|
92 | $stmt->execute(); |
|
93 | $results = $stmt->fetchAll(); |
|
94 | ||
95 | $storedEvents = array_map(function($event) { |
|
96 | return new StoredEvent( |
|
97 | $event['id'], |
|
98 | $event['stream_id'], |
|
99 | $event['type'], |
|
100 | $event['event'], |
|
101 | $event['metadata'], |
|
102 | new \DateTimeImmutable($event['occurred_on']), |
|
103 | Version::fromString($event['version']) |
|
104 | ); |
|
105 | }, $results); |
|
106 | ||
107 | return $this->domainEventStreamFromStoredEvents($storedEvents); |
|
108 | } |
|
109 | ||
110 | /** |
|
111 | * @return EventStreamInterface[] |
|
@@ 131-153 (lines=23) @@ | ||
128 | /** |
|
129 | * @return EventStreamInterface |
|
130 | */ |
|
131 | public function readAllEvents() |
|
132 | { |
|
133 | $stmt = $this->connection->prepare( |
|
134 | 'SELECT * |
|
135 | FROM events' |
|
136 | ); |
|
137 | $stmt->execute(); |
|
138 | $results = $stmt->fetchAll(); |
|
139 | ||
140 | $storedEvents = array_map(function($event) { |
|
141 | return new StoredEvent( |
|
142 | $event['id'], |
|
143 | $event['stream_id'], |
|
144 | $event['type'], |
|
145 | $event['event'], |
|
146 | $event['metadata'], |
|
147 | new \DateTimeImmutable($event['occurred_on']), |
|
148 | Version::fromString($event['version']) |
|
149 | ); |
|
150 | }, $results); |
|
151 | ||
152 | return $this->domainEventStreamFromStoredEvents($storedEvents); |
|
153 | } |
|
154 | ||
155 | /** |
|
156 | * @param string $streamId |
|
@@ 228-254 (lines=27) @@ | ||
225 | * @param Version $version |
|
226 | * @return EventStreamInterface |
|
227 | */ |
|
228 | protected function readStoredEventsOfTypeAndVersion($type, $version) |
|
229 | { |
|
230 | $stmt = $this->connection->prepare( |
|
231 | 'SELECT * |
|
232 | FROM events |
|
233 | WHERE type = :type |
|
234 | AND version = :version' |
|
235 | ); |
|
236 | $stmt->bindValue(':type', $type); |
|
237 | $stmt->bindValue(':version', $version); |
|
238 | $stmt->execute(); |
|
239 | $results = $stmt->fetchAll(); |
|
240 | ||
241 | $storedEvents = array_map(function($result) { |
|
242 | return new StoredEvent( |
|
243 | $result['id'], |
|
244 | $result['stream_id'], |
|
245 | $result['type'], |
|
246 | $result['event'], |
|
247 | $result['metadata'], |
|
248 | new \DateTimeImmutable($result['occurred_on']), |
|
249 | Version::fromString($result['version']) |
|
250 | ); |
|
251 | }, $results); |
|
252 | ||
253 | return new EventStream($storedEvents); |
|
254 | } |
|
255 | ||
256 | public function initialize() |
|
257 | { |