@@ -43,7 +43,18 @@ object SiriusPersistenceActor {
4343 * @param end last sequence number of the range, inclusive
4444 */
4545 case class GetLogSubrange (begin : Long , end : Long ) extends LogQuery
46- case class GetLogSubrangeToLimit (begin : Long , events : Long ) extends LogQuery
46+ /**
47+ * Message for directly requesting a chunk of the log from a node.
48+ *
49+ * SiriusPersistenceActor is expected to reply with a LogSubrange
50+ * when receiving this message. This range should be as complete
51+ * as possible.
52+ *
53+ * @param begin first sequence number of the range
54+ * @param end last sequence number of the range, inclusive
55+ * @param limit the maximum number of events
56+ */
57+ case class GetLogSubrangeWithLimit (begin : Long , end : Long , limit : Long ) extends LogQuery
4758
4859 trait LogSubrange
4960 trait PopulatedSubrange extends LogSubrange {
@@ -132,34 +143,11 @@ class SiriusPersistenceActor(stateActor: ActorRef,
132143
133144 lastWriteTime = thisWriteTime
134145
135- case GetLogSubrangeToLimit (start, limit) =>
136- val buffer = siriusLog.foldLeftWhile(start)(ListBuffer .empty[OrderedEvent ])(buffer => buffer.length < limit)(
137- (acc, event) => acc += event
138- )
139- if (buffer.isEmpty) {
140- sender ! EmptySubrange
141- }
142- if (buffer.length < limit) {
143- sender ! PartialSubrange (buffer.head.sequence, buffer.last.sequence, buffer.toList)
144- } else {
145- sender ! CompleteSubrange (buffer.head.sequence, buffer.last.sequence, buffer.toList)
146- }
147-
148- case GetLogSubrange (rangeStart, rangeEnd) if rangeEnd < siriusLog.getNextSeq => // we can answer fully
149- val events = siriusLog.foldLeftRange(rangeStart, rangeEnd)(ListBuffer .empty[OrderedEvent ])(
150- (acc, event) => acc += event
151- )
152- sender ! CompleteSubrange (rangeStart, rangeEnd, events.toList)
153-
154- case GetLogSubrange (rangeStart, _) if siriusLog.getNextSeq <= rangeStart => // we can't send anything useful back
155- sender ! EmptySubrange
146+ case GetLogSubrangeWithLimit (start, end, limit) =>
147+ sender ! queryLimitedSubrange(start, end, limit)
156148
157- case GetLogSubrange (rangeStart, _) => // we can respond partially
158- val lastSeq = siriusLog.getNextSeq - 1
159- val events = siriusLog.foldLeftRange(rangeStart, lastSeq)(ListBuffer .empty[OrderedEvent ])(
160- (acc, event) => acc += event
161- )
162- sender ! PartialSubrange (rangeStart, lastSeq, events.toList)
149+ case GetLogSubrange (start, end) =>
150+ sender ! querySubrange(start, end)
163151
164152 case GetNextLogSeq =>
165153 sender ! siriusLog.getNextSeq
@@ -172,6 +160,72 @@ class SiriusPersistenceActor(stateActor: ActorRef,
172160 case _ : SiriusResult =>
173161 }
174162
163+ private def queryLimitedSubrange (rangeStart : Long , rangeEnd : Long , limit : Long ): LogSubrange =
164+ if (limit <= 0 || rangeEnd < rangeStart || rangeEnd <= 0 ) {
165+ // invalid query subrange or limit, we can't send anything useful back
166+ EmptySubrange
167+ } else if (limit > (rangeEnd - rangeStart)) {
168+ // limit is larger than the subrange window, no need to enforce limit
169+ querySubrange(rangeStart, rangeEnd)
170+ } else {
171+ val nextSeq = siriusLog.getNextSeq
172+ if (rangeStart >= nextSeq) {
173+ // query is out of range, we can't send anything useful back
174+ EmptySubrange
175+ } else if (rangeEnd >= nextSeq) {
176+ // we can only answer partially
177+ val lastSeq = nextSeq - 1
178+ val buffer = siriusLog.foldLeftRangeWhile(rangeStart, lastSeq)(ListBuffer .empty[OrderedEvent ])(
179+ // continue folding events as long as the buffer is smaller than the limit
180+ buffer => buffer.size < limit
181+ )(
182+ (acc, event) => acc += event
183+ )
184+ if (buffer.size < limit) {
185+ PartialSubrange (rangeStart, lastSeq, buffer.toList)
186+ } else {
187+ PartialSubrange (rangeStart, buffer.last.sequence, buffer.toList)
188+ }
189+ } else {
190+ val buffer = siriusLog.foldLeftRangeWhile(rangeStart, rangeEnd)(ListBuffer .empty[OrderedEvent ])(
191+ // continue folding events as long as the buffer is smaller than the limit
192+ buffer => buffer.size < limit
193+ )(
194+ (acc, event) => acc += event
195+ )
196+ if (buffer.size < limit) {
197+ CompleteSubrange (rangeStart, rangeEnd, buffer.toList)
198+ } else {
199+ PartialSubrange (rangeStart, buffer.last.sequence, buffer.toList)
200+ }
201+ }
202+ }
203+
204+ private def querySubrange (rangeStart : Long , rangeEnd : Long ): LogSubrange =
205+ if (rangeEnd < rangeStart || rangeEnd <= 0 ) {
206+ // invalid query subrange or limit, we can't send anything useful back
207+ EmptySubrange
208+ } else {
209+ val nextSeq = siriusLog.getNextSeq
210+ if (rangeStart >= nextSeq) {
211+ // query is out of range, we can't send anything useful back
212+ EmptySubrange
213+ } else if (rangeEnd >= nextSeq) {
214+ // we can answer partially
215+ val lastSeq = nextSeq - 1
216+ val buffer = siriusLog.foldLeftRange(rangeStart, lastSeq)(ListBuffer .empty[OrderedEvent ])(
217+ (acc, event) => acc += event
218+ )
219+ PartialSubrange (rangeStart, lastSeq, buffer.toList)
220+ } else {
221+ // we can answer fully
222+ val buffer = siriusLog.foldLeftRange(rangeStart, rangeEnd)(ListBuffer .empty[OrderedEvent ])(
223+ (acc, event) => acc += event
224+ )
225+ CompleteSubrange (rangeStart, rangeEnd, buffer.toList)
226+ }
227+ }
228+
175229 /**
176230 * Monitoring hooks
177231 */
0 commit comments