sandbox-on-x: Support beginAfter in state updates. (#9624)

* sandbox-on-x: Support `beginAfter` in state updates. CHANGELOG_BEGIN CHANGELOG_END * sandbox-on-x: Update a TODO comment.
digital-asset · May 10, 2021 · 6c919b3 · 6c919b3
1 parent 26a8011
commit 6c919b3
Showing 1 changed file with 7 additions and 8 deletions.
diff --git a/ledger/sandbox-on-x/src/main/scala/com/daml/ledger/sandbox/ReadWriteServiceBridge.scala b/ledger/sandbox-on-x/src/main/scala/com/daml/ledger/sandbox/ReadWriteServiceBridge.scala
@@ -112,20 +112,19 @@ case class ReadWriteServiceBridge(
   var stateUpdatesWasCalledAlready = false
   override def stateUpdates(beginAfter: Option[Offset]): Source[(Offset, Update), NotUsed] = {
     // TODO for PoC purposes:
-    //   no beginAfter supported
-    //   neither multiple subscriptions
-    //   neither bootstrapping the bridge from indexer persistence
-    assert(
-      beginAfter.isEmpty,
-      "Re-subscribing not supported. Only supported to subscribe once, and from inception.",
-    )
+    //   This method may only be called once, either with `beginAfter` set or unset.
+    //   A second call will result in an error unless the server is restarted.
+    //   Bootstrapping the bridge from indexer persistence is supported.
     synchronized {
       if (stateUpdatesWasCalledAlready)
         throw new IllegalStateException("not allowed to call this twice")
       else stateUpdatesWasCalledAlready = true
     }
     logger.info("Indexer subscribed to state updates.")
-    queueSource
+    beginAfter match {
+      case None => queueSource
+      case Some(offset) => queueSource.dropWhile(_._1 <= offset)
+    }
   }
 
   val (queue: BoundedSourceQueue[Submission], queueSource: Source[(Offset, Update), NotUsed]) =