utility for simplifying FanOutShape2s; use in ContractsService

digital-asset · S11001001 · Jan 15, 2020 · Dec 18, 2019 · Dec 18, 2019 · Dec 19, 2019
commit 6f4e81bbdd31ee03ef97baf62e220621959f6087
diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsFetch.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsFetch.scala
@@ -9,13 +9,14 @@ import akka.stream.scaladsl.{
   Concat,
   Flow,
   GraphDSL,
+  Keep,
   Partition,
   RunnableGraph,
   Sink,
   SinkQueueWithCancel,
   Source
 }
-import akka.stream.{ClosedShape, FanOutShape2, Graph, Materializer}
+import akka.stream.{ClosedShape, FanOutShape2, FlowShape, Graph, Materializer}
 import com.digitalasset.http.Statement.discard
 import com.digitalasset.http.dbbackend.ContractDao.StaleOffsetException
 import com.digitalasset.http.dbbackend.{ContractDao, Queries}
@@ -141,7 +142,7 @@ private class ContractsFetch(
       GraphDSL.create(
         Sink.queue[ConnectionIO[Unit]](),
         Sink.last[OffsetBookmark[String]]
-      )((a, b) => (a, b)) { implicit builder => (acsSink, offsetSink) =>
+      )(Keep.both) { implicit builder => (acsSink, offsetSink) =>
         import GraphDSL.Implicits._
 
         val txnK = getCreatesAndArchivesSince(
@@ -260,6 +261,17 @@ private[http] object ContractsFetch {
     InsertDeleteStep(csb.result() filter (ce => !as.contains(ce.contractId)), as)
   }
 
+  /** Several of the graphs here have a second output guaranteed to deliver only one value.
+    * This turns such a graph into a flow with the value materialized.
+    */
+  def matSecondOut[A, Y, Z](g: Graph[FanOutShape2[A, Y, Z], NotUsed]): Flow[A, Y, Future[Z]] =
+    Flow fromGraph GraphDSL.create(Sink.head[Z]) { implicit b => zOut =>
+      import GraphDSL.Implicits._
+      val gs = b add g
+      gs.out1 ~> zOut
+      new FlowShape(gs.in, gs.out0)
+    }
+
   /** Like `acsAndBoundary`, but also include the events produced by `transactionsSince`
     * after the ACS's last offset, terminating with the last offset of the last transaction,
     * or the ACS's last offset if there were no transactions.

diff --git a/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala b/ledger-service/http-json/src/main/scala/com/digitalasset/http/ContractsService.scala
@@ -5,9 +5,9 @@ package com.digitalasset.http
 
 import akka.NotUsed
 import akka.stream.scaladsl._
-import akka.stream.{Materializer, SourceShape}
+import akka.stream.Materializer
 import com.digitalasset.daml.lf
-import com.digitalasset.http.ContractsFetch.{InsertDeleteStep, OffsetBookmark}
+import com.digitalasset.http.ContractsFetch.InsertDeleteStep
 import com.digitalasset.http.dbbackend.ContractDao
 import com.digitalasset.http.domain.{GetActiveContractsRequest, JwtPayload, TemplateId}
 import com.digitalasset.http.json.JsonProtocol.LfValueCodec
@@ -245,29 +245,23 @@ class ContractsService(
       templateId: domain.TemplateId.RequiredPkg)
     : Source[InsertDeleteStep[api.event.CreatedEvent], NotUsed] = {
 
-    val graph = GraphDSL.create() { implicit b =>
-      import GraphDSL.Implicits._
-
-      val source = getActiveContracts(jwt, transactionFilter(party, templateId), true)
-
-      val transactionsSince
-        : api.ledger_offset.LedgerOffset => Source[api.transaction.Transaction, NotUsed] =
-        getCreatesAndArchivesSince(
-          jwt,
-          transactionFilter(party, templateId),
-          _: api.ledger_offset.LedgerOffset)
-
-      val contractsAndBoundary = b add ContractsFetch.acsFollowingAndBoundary(transactionsSince)
-      val offsetSink = b add Sink.foreach[OffsetBookmark[String]] { a =>
-        logger.debug(s"contracts fetch completed at: ${a.toString}")
+    val source = getActiveContracts(jwt, transactionFilter(party, templateId), true)
+
+    val transactionsSince
+      : api.ledger_offset.LedgerOffset => Source[api.transaction.Transaction, NotUsed] =
+      getCreatesAndArchivesSince(
+        jwt,
+        transactionFilter(party, templateId),
+        _: api.ledger_offset.LedgerOffset)
+
+    import ContractsFetch.{acsFollowingAndBoundary, matSecondOut}
+    val contractsAndBoundary = matSecondOut(acsFollowingAndBoundary(transactionsSince))
+    source
+      .viaMat(contractsAndBoundary)(Keep.right)
+      .mapMaterializedValue { fob =>
+        fob.foreach(a => logger.debug(s"contracts fetch completed at: ${a.toString}"))
+        NotUsed
       }
-
-      source ~> contractsAndBoundary.in
-      contractsAndBoundary.out1 ~> offsetSink
-      new SourceShape(contractsAndBoundary.out0)
-    }
-
-    Source.fromGraph(graph)
   }
 
   @SuppressWarnings(Array("org.wartremover.warts.Any"))