Make end sequence optional

Author: HaloFour

src/main/scala/com/comcast/xfinity/sirius/api/impl/state/SiriusPersistenceActor.scala

@@ -54,7 +54,7 @@ object SiriusPersistenceActor {
    * @param end last sequence number of the range, inclusive
    * @param limit the maximum number of events
    */
-  case class GetLogSubrangeWithLimit(begin: Long, end: Long, limit: Long) extends LogQuery
+  case class GetLogSubrangeWithLimit(begin: Long, end: Option[Long], limit: Long) extends LogQuery
 
   trait LogSubrange
   trait PopulatedSubrange extends LogSubrange {
@@ -143,9 +143,12 @@ class SiriusPersistenceActor(stateActor: ActorRef,
 
       lastWriteTime = thisWriteTime
 
-    case GetLogSubrangeWithLimit(start, end, limit) =>
+    case GetLogSubrangeWithLimit(start, Some(end), limit) =>
       sender ! queryLimitedSubrange(start, end, limit)
 
+    case GetLogSubrangeWithLimit(start, None, limit) =>
+      sender ! queryLimited(start, limit)
+
     case GetLogSubrange(start, end) =>
       sender ! querySubrange(start, end)
 
@@ -201,6 +204,33 @@ class SiriusPersistenceActor(stateActor: ActorRef,
       }
     }
 
+  private def queryLimited(rangeStart: Long, limit: Long): LogSubrange = {
+    if (limit <= 0) {
+      // invalid query subrange or limit, we can't send anything useful back
+      EmptySubrange
+    } else {
+      val nextSeq = siriusLog.getNextSeq
+      if (rangeStart >= nextSeq) {
+        // query is out of range, we can't send anything useful back
+        EmptySubrange
+      } else {
+        val nextSeq = siriusLog.getNextSeq
+        val lastSeq = nextSeq - 1
+        val buffer = siriusLog.foldLeftRangeWhile(rangeStart, lastSeq)(ListBuffer.empty[OrderedEvent])(
+          // continue folding events as long as the buffer is smaller than the limit
+          buffer => buffer.size < limit
+        )(
+          (acc, event) => acc += event
+        )
+        if (buffer.size < limit) {
+          CompleteSubrange(rangeStart, lastSeq, buffer.toList)
+        } else {
+          PartialSubrange(rangeStart, buffer.last.sequence, buffer.toList)
+        }
+      }
+    }
+  }
+
   private def querySubrange(rangeStart: Long, rangeEnd: Long): LogSubrange =
     if (rangeEnd < rangeStart || rangeEnd <= 0) {
       // invalid query subrange or limit, we can't send anything useful back

src/test/scala/com/comcast/xfinity/sirius/api/impl/state/SiriusPersistenceActorTest.scala

@@ -194,7 +194,7 @@ class SiriusPersistenceActorTest extends NiceTest {
           val mockLog = makeMockLog(ListBuffer(event1, event2), 10L)
           val underTest = makePersistenceActor(siriusLog = mockLog)
 
-          senderProbe.send(underTest, GetLogSubrangeWithLimit(1, 2, 2))
+          senderProbe.send(underTest, GetLogSubrangeWithLimit(1, Some(2), 2))
 
           verifyFoldLeftRanged(mockLog, 1, 2)
           senderProbe.expectMsg(CompleteSubrange(1, 2, List(event1, event2)))
@@ -205,33 +205,33 @@ class SiriusPersistenceActorTest extends NiceTest {
           val senderProbe = TestProbe()
 
           val event1 = mock[OrderedEvent]
-          doReturn(8L).when(event1).sequence
+          doReturn(1L).when(event1).sequence
           val event2 = mock[OrderedEvent]
-          doReturn(9L).when(event2).sequence
+          doReturn(2L).when(event2).sequence
           val mockLog = makeMockLog(ListBuffer(event1, event2), 10L)
           val underTest = makePersistenceActor(siriusLog = mockLog)
 
-          senderProbe.send(underTest, GetLogSubrangeWithLimit(8, Long.MaxValue, 3))
+          senderProbe.send(underTest, GetLogSubrangeWithLimit(1, None, 3))
 
-          verifyFoldLeftWhile(mockLog, 8, 9)
-          senderProbe.expectMsg(PartialSubrange(8, 9, List(event1, event2)))
+          verifyFoldLeftWhile(mockLog, 1, 9)
+          senderProbe.expectMsg(CompleteSubrange(1, 9, List(event1, event2)))
         }
       }
       describe("when we can partially reply due to limit") {
         it("should build the list of events and reply with it") {
           val senderProbe = TestProbe()
 
           val event1 = mock[OrderedEvent]
-          doReturn(8L).when(event1).sequence
+          doReturn(1L).when(event1).sequence
           val event2 = mock[OrderedEvent]
-          doReturn(9L).when(event2).sequence
+          doReturn(2L).when(event2).sequence
           val mockLog = makeMockLog(ListBuffer(event1, event2), 11L)
           val underTest = makePersistenceActor(siriusLog = mockLog)
 
-          senderProbe.send(underTest, GetLogSubrangeWithLimit(8, 10, 2))
+          senderProbe.send(underTest, GetLogSubrangeWithLimit(1, Some(10), 2))
 
-          verifyFoldLeftWhile(mockLog, 8, 10)
-          senderProbe.expectMsg(PartialSubrange(8, 9, List(event1, event2)))
+          verifyFoldLeftWhile(mockLog, 1, 10)
+          senderProbe.expectMsg(PartialSubrange(1, 2, List(event1, event2)))
         }
       }
       describe("when we can't send anything useful at all") {
@@ -241,7 +241,7 @@ class SiriusPersistenceActorTest extends NiceTest {
           val mockLog = makeMockLog(ListBuffer(), 5L)
           val underTest = makePersistenceActor(siriusLog = mockLog)
 
-          senderProbe.send(underTest, GetLogSubrangeWithLimit(8, Long.MaxValue, 11))
+          senderProbe.send(underTest, GetLogSubrangeWithLimit(8, None, 11))
 
           senderProbe.expectMsg(EmptySubrange)
         }