Add GetLogRangeLimit message to read log ranges by limit of events

Author: HaloFour

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

@@ -15,12 +15,14 @@
  */
 package com.comcast.xfinity.sirius.api.impl.state
 
-import akka.actor.{Props, Actor, ActorRef}
+import akka.actor.{Actor, ActorRef, Props}
 import com.comcast.xfinity.sirius.writeaheadlog.SiriusLog
 import com.comcast.xfinity.sirius.api.{SiriusConfiguration, SiriusResult}
 import com.comcast.xfinity.sirius.api.impl._
 import com.comcast.xfinity.sirius.admin.MonitoringHooks
 
+import scala.collection.mutable.ListBuffer
+
 object SiriusPersistenceActor {
 
   /**
@@ -41,6 +43,7 @@ object SiriusPersistenceActor {
    * @param end last sequence number of the range, inclusive
    */
   case class GetLogSubrange(begin: Long, end: Long) extends LogQuery
+  case class GetLogRangeLimit(begin: Long, events: Long) extends LogQuery
 
   trait LogSubrange
   trait PopulatedSubrange extends LogSubrange {
@@ -129,21 +132,34 @@ class SiriusPersistenceActor(stateActor: ActorRef,
 
       lastWriteTime = thisWriteTime
 
+    case GetLogRangeLimit(start, limit) =>
+      val buffer = siriusLog.foldLeftWhile(start)(ListBuffer.empty[OrderedEvent])(buffer => buffer.length < limit)(
+        (acc, event) => acc += event
+      )
+      if (buffer.isEmpty) {
+        sender ! EmptySubrange
+      }
+      if (buffer.length < limit) {
+        sender ! PartialSubrange(buffer.head.sequence, buffer.last.sequence, buffer.toList)
+      } else {
+        sender ! CompleteSubrange(buffer.head.sequence, buffer.last.sequence, buffer.toList)
+      }
+
     case GetLogSubrange(rangeStart, rangeEnd) if rangeEnd < siriusLog.getNextSeq => // we can answer fully
-      val events = siriusLog.foldLeftRange(rangeStart, rangeEnd)(List[OrderedEvent]())(
-        (acc, event) => event :: acc
-      ).reverse
-      sender ! CompleteSubrange(rangeStart, rangeEnd, events)
+      val events = siriusLog.foldLeftRange(rangeStart, rangeEnd)(ListBuffer.empty[OrderedEvent])(
+        (acc, event) => acc += event
+      )
+      sender ! CompleteSubrange(rangeStart, rangeEnd, events.toList)
 
-    case GetLogSubrange(rangeStart, rangeEnd) if siriusLog.getNextSeq <= rangeStart => // we can't send anything useful back
+    case GetLogSubrange(rangeStart, _) if siriusLog.getNextSeq <= rangeStart => // we can't send anything useful back
       sender ! EmptySubrange
 
-    case GetLogSubrange(rangeStart, rangeEnd) => // we can respond partially
+    case GetLogSubrange(rangeStart, _) => // we can respond partially
       val lastSeq = siriusLog.getNextSeq - 1
-      val events = siriusLog.foldLeftRange(rangeStart, lastSeq)(List[OrderedEvent]())(
-        (acc, event) => event :: acc
-      ).reverse
-      sender ! PartialSubrange(rangeStart, lastSeq, events)
+      val events = siriusLog.foldLeftRange(rangeStart, lastSeq)(ListBuffer.empty[OrderedEvent])(
+        (acc, event) => acc += event
+      )
+      sender ! PartialSubrange(rangeStart, lastSeq, events.toList)
 
     case GetNextLogSeq =>
       sender ! siriusLog.getNextSeq

src/main/scala/com/comcast/xfinity/sirius/uberstore/UberPair.scala

@@ -107,6 +107,11 @@ class UberPair(dataFile: UberDataFile, index: SeqIndex) {
     )
   }
 
+  def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T = {
+    val (startOffset, _) = index.getOffsetRange(startSeq, Long.MaxValue)
+    dataFile.foldLeftWhile(startOffset)(acc0)(pred)(foldFun)
+  }
+
   /**
    * Close underlying file handles or connections.  This UberStoreFilePair should not be used after
    * close is called.

src/main/scala/com/comcast/xfinity/sirius/uberstore/UberStore.scala

@@ -82,6 +82,13 @@ class UberStore private[uberstore] (baseDir: String, uberpair: UberPair) extends
   def foldLeftRange[T](startSeq: Long, endSeq: Long)(acc0: T)(foldFun: (T, OrderedEvent) => T): T =
     uberpair.foldLeftRange(startSeq, endSeq)(acc0)(foldFun)
 
+  /**
+   * @inheritdoc
+   */
+  def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T = {
+    uberpair.foldLeftWhile(startSeq)(acc0)(pred)(foldFun)
+  }
+
   /**
    * Close underlying file handles or connections.  This UberStore should not be used after
    * close is called.

src/main/scala/com/comcast/xfinity/sirius/uberstore/data/UberDataFile.scala

@@ -101,15 +101,24 @@ private[uberstore] class UberDataFile(dataFileName: String,
   def foldLeftRange[T](baseOff: Long, endOff: Long)(acc0: T)(foldFun: (T, Long, OrderedEvent) => T): T = {
     val readHandle = fileHandleFactory.createReadHandle(dataFileName, baseOff)
     try {
-      foldLeftUntil(readHandle, endOff, acc0, foldFun)
+      foldLeftUntilOffset(readHandle, endOff, acc0, foldFun)
+    } finally {
+      readHandle.close()
+    }
+  }
+
+  def foldLeftWhile[T](baseOff:Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T = {
+    val readHandle = fileHandleFactory.createReadHandle(dataFileName, baseOff)
+    try {
+      foldLeftWhile(readHandle, acc0, pred, foldFun)
     } finally {
       readHandle.close()
     }
   }
 
   // private low low low level fold left
   @tailrec
-  private def foldLeftUntil[T](readHandle: UberDataFileReadHandle, maxOffset: Long, acc: T, foldFun: (T, Long, OrderedEvent) => T): T = {
+  private def foldLeftUntilOffset[T](readHandle: UberDataFileReadHandle, maxOffset: Long, acc: T, foldFun: (T, Long, OrderedEvent) => T): T = {
     val offset = readHandle.offset()
     if (offset > maxOffset) {
       acc
@@ -118,7 +127,21 @@ private[uberstore] class UberDataFile(dataFileName: String,
         case None => acc
         case Some(bytes) =>
           val accNew = foldFun(acc, offset, codec.deserialize(bytes))
-          foldLeftUntil(readHandle, maxOffset, accNew, foldFun)
+          foldLeftUntilOffset(readHandle, maxOffset, accNew, foldFun)
+      }
+    }
+  }
+
+  @tailrec
+  private def foldLeftWhile[T](readHandle: UberDataFileReadHandle, acc: T, pred: T => Boolean, foldFun: (T, OrderedEvent) => T): T = {
+    if (!pred(acc)) {
+      acc
+    } else {
+      fileOps.readNext(readHandle) match {
+        case None => acc
+        case Some(bytes) =>
+          val accNew = foldFun(acc, codec.deserialize(bytes))
+          foldLeftWhile(readHandle, accNew, pred, foldFun)
       }
     }
   }

src/main/scala/com/comcast/xfinity/sirius/uberstore/segmented/Segment.scala

@@ -107,7 +107,7 @@ class Segment private[uberstore](val location: File,
    * Get the number of entries written to the Segment
    * @return number of entries in the Segment
    */
-  def size = index.size
+  def size: Long = index.size
 
   /**
    * Write OrderedEvent event into this dir. Will fail if closed or sequence
@@ -136,10 +136,15 @@ class Segment private[uberstore](val location: File,
       (acc, event) => acc + event.request.key
     )
 
+  /**
+   * Get the first sequence number in this dir.
+   */
+  def getFirstSeq: Option[Long] = index.getMinSeq
+
   /**
    * Get the next possible sequence number in this dir.
    */
-  def getNextSeq = index.getMaxSeq match {
+  def getNextSeq: Long = index.getMaxSeq match {
     case None => 1L
     case Some(seq) => seq + 1
   }
@@ -167,6 +172,13 @@ class Segment private[uberstore](val location: File,
     )
   }
 
+  def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T = {
+    val (startOffset, _) = index.getOffsetRange(startSeq, Long.MaxValue)
+    dataFile.foldLeftWhile(startOffset)(acc0)(pred)(
+      (acc, evt) => foldFun(acc, evt)
+    )
+  }
+
   /**
    * Close underlying file handles or connections.  This Segment should not be used after
    * close is called.

src/main/scala/com/comcast/xfinity/sirius/uberstore/segmented/SegmentedUberStore.scala

@@ -151,6 +151,15 @@ class SegmentedUberStore private[segmented] (base: JFile,
     liveDir.foldLeftRange(startSeq, endSeq)(res0)(foldFun)
   }
 
+  /**
+   * @inheritdoc
+   */
+  def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T = {
+    val res0 = readOnlyDirs.foldLeft(acc0)(
+      (acc, dir) => dir.foldLeftWhile(startSeq)(acc)(pred)(foldFun)
+    )
+    liveDir.foldLeftWhile(startSeq)(res0)(pred)(foldFun)
+  }
 
   /**
    * Close underlying file handles or connections.  This SegmentedUberStore should not be used after

src/main/scala/com/comcast/xfinity/sirius/uberstore/seqindex/DiskOnlySeqIndex.scala

@@ -51,7 +51,17 @@ class DiskOnlySeqIndex private(handle: RandomAccessFile,
   var isClosed = false
   var size: Long = handle.length() / 24
 
-  var maxSeq = {
+  private var minSeq = {
+    if (handle.length == 0)
+      None
+    else {
+      handle.seek(0)
+      val (seq, _) = fileOps.readEntry(handle)
+      Some(seq)
+    }
+  }
+
+  private var maxSeq = {
     if (handle.length == 0)
       None
     else {
@@ -87,6 +97,11 @@ class DiskOnlySeqIndex private(handle: RandomAccessFile,
     getOffsetForAux(0, handle.length)
   }
 
+  /**
+   * {@inheritdoc}
+   */
+  def getMinSeq: Option[Long] = minSeq
+
   /**
    * {@inheritdoc}
    */
@@ -98,6 +113,9 @@ class DiskOnlySeqIndex private(handle: RandomAccessFile,
   def put(seq: Long, offset: Long): Unit = synchronized {
     handle.seek(handle.length)
     fileOps.put(handle, seq, offset)
+    if (minSeq.isEmpty) {
+      minSeq = Some(seq)
+    }
     maxSeq = Some(seq)
     size += 1
   }

src/main/scala/com/comcast/xfinity/sirius/uberstore/seqindex/SeqIndex.scala

@@ -25,6 +25,13 @@ trait SeqIndex {
    */
   def getOffsetFor(seq: Long): Option[Long]
 
+  /**
+   * Get the minimum sequence number stored, if such exists
+   *
+   * @return Some(sequence) or None if none such exists
+   */
+  def getMinSeq: Option[Long]
+
   /**
    * Get the maximum sequence number stored, if such exists
    *

src/main/scala/com/comcast/xfinity/sirius/writeaheadlog/CachedSiriusLog.scala

@@ -108,6 +108,9 @@ class CachedSiriusLog(log: SiriusLog, maxCacheSize: Int) extends SiriusLog {
       .values.asScala.foldLeft(acc0)(foldFun)
   }
 
+  override def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T =
+    log.foldLeftWhile(startSeq)(acc0)(pred)(foldFun)
+
   def getNextSeq = log.getNextSeq
 
   def compact(): Unit = {

src/main/scala/com/comcast/xfinity/sirius/writeaheadlog/SiriusLog.scala

@@ -59,6 +59,15 @@ trait SiriusLog {
    */
   def foldLeftRange[T](startSeq: Long, endSeq: Long)(acc0: T)(foldFun: (T, OrderedEvent) => T): T
 
+  /**
+   * Fold left across log entries while a condition is met
+   * @param startSeq sequence number to start with, inclusive
+   * @param acc0 initial accumulator value
+   * @param pred condition to continue accumulating log entries
+   * @param foldFun function to apply to the log entry, the result being the new accumulator
+   */
+  def foldLeftWhile[T](startSeq: Long)(acc0: T)(pred: T => Boolean)(foldFun: (T, OrderedEvent) => T): T
+
   /**
    * retrieves the next sequence number to be written
    */