extractor: subscribe to multiple parties

docs/source/support/release-notes.rst

@@ -18,6 +18,14 @@ DAML Assistant
 
 - You can now see all available versions with ``daml version`` using the ``--all`` flag.
 
+SQL Extractor
+~~~~~~~~~~~~~
+
+- The extractor ``--party`` option may now specify multiple parties, separated by commas;
+  e.g. instead of ``--party Bob`` you can say ``--party Bob,Bar,Baz`` and get the contracts
+  for all three parties in the database.
+  See `#1360 <https://github.com/digital-asset/daml/pull/1360>`__.
+
 0.12.20 - 2019-05-23
 --------------------
 

extractor/BUILD.bazel

@@ -41,23 +41,20 @@ compileDependencies = [
     "//3rdparty/jvm/io/grpc:grpc_netty",
 ]
 
-daml_compile(
-    name = "RecordsAndVariants",
-    main_src = "src/test/resources/damls/RecordsAndVariants.daml",
-    target = "1.3",
-)
-
-daml_compile(
-    name = "PrimitiveTypes",
-    main_src = "src/test/resources/damls/PrimitiveTypes.daml",
-    target = "1.3",
-)
+TEST_DARS = [
+    "RecordsAndVariants",
+    "PrimitiveTypes",
+    "TransactionExample",
+]
 
-daml_compile(
-    name = "TransactionExample",
-    main_src = "src/test/resources/damls/TransactionExample.daml",
-    target = "1.3",
-)
+[
+    daml_compile(
+        name = darmod,
+        main_src = "src/test/resources/damls/%s.daml" % darmod,
+        target = "1.3",
+    )
+    for darmod in TEST_DARS
+]
 
 testDependencies = [
     ":extractor",

extractor/src/main/scala/com/digitalasset/extractor/Extractor.scala

@@ -22,6 +22,7 @@ import com.digitalasset.ledger.client.LedgerClient
 import com.digitalasset.ledger.client.configuration._
 
 import io.grpc.netty.{NegotiationType, NettyChannelBuilder}
+import scala.collection.breakOut
 import scala.concurrent.duration._
 import scala.concurrent.Future
 import scala.util.control.NonFatal
@@ -110,6 +111,11 @@ class Extractor[T <: Target](config: ExtractorConfig, target: T) {
     } yield ()
   }
 
+  private def selectTransactions: TransactionFilter = {
+    val templateSelection = Filters.defaultInstance
+    TransactionFilter(config.parties.toList.map(_ -> templateSelection)(breakOut))
+  }
+
   private def streamTransactions(
       client: LedgerClient,
       writer: Writer,
@@ -126,7 +132,7 @@ class Extractor[T <: Target](config: ExtractorConfig, target: T) {
           .getTransactionTrees(
             LedgerOffset(startOffSet),
             streamUntil,
-            TransactionFilter(Map(config.party -> Filters.defaultInstance)),
+            selectTransactions,
             verbose = true
           )
           .via(killSwitch.flow)

extractor/src/main/scala/com/digitalasset/extractor/config/ConfigParser.scala

@@ -279,7 +279,7 @@ object ConfigParser {
         cliParams.ledgerPort,
         from,
         to,
-        cliParams.party,
+        ExtractorConfig.parties(cliParams.party),
         tlsConfig
       )
 

extractor/src/main/scala/com/digitalasset/extractor/config/ExtractorConfig.scala

@@ -4,6 +4,10 @@
 package com.digitalasset.extractor.config
 
 import java.util.UUID
+import scalaz.OneAnd
+import scalaz.syntax.foldable._
+import scalaz.std.list._
+import scalaz.std.string._
 
 import com.digitalasset.ledger.api.v1.ledger_offset.LedgerOffset
 import com.digitalasset.ledger.api.tls.TlsConfiguration
@@ -20,7 +24,17 @@ final case class ExtractorConfig(
     ledgerPort: Int,
     from: LedgerOffset,
     to: SnapshotEndSetting,
-    party: String,
+    parties: ExtractorConfig.Parties,
     tlsConfig: TlsConfiguration,
     appId: String = s"Extractor-${UUID.randomUUID().toString}"
-)
+) {
+  def partySpec: String = parties intercalate ","
+}
+
+object ExtractorConfig {
+  type Parties = OneAnd[List, String]
+  def parties(spec: String): Parties = {
+    val Array(hd, tl @ _*) = spec split ','
+    OneAnd(hd, tl.toList)
+  }
+}

extractor/src/main/scala/com/digitalasset/extractor/writers/postgresql/StateHandler.scala

@@ -102,7 +102,7 @@ object StateHandler {
       target.copy(connectUrl = "**masked**", user = "**masked**", password = "**masked**"),
       config.from,
       config.to,
-      config.party
+      config.partySpec
     )
   }
 
@@ -133,7 +133,7 @@ object StateHandler {
       )
       _ <- validateParam(
         previousStatus.startUpParameters.party,
-        config.party,
+        config.partySpec,
         "`--party`"
       )
       _ <- validateParam(

extractor/src/test/lib/scala/com/digitalasset/extractor/services/ExtractorFixture.scala

@@ -11,6 +11,7 @@ import com.digitalasset.ledger.api.tls.TlsConfiguration
 import com.digitalasset.platform.sandbox.persistence.PostgresAround
 import com.digitalasset.platform.sandbox.services.SandboxFixture
 
+import scalaz.OneAnd
 import cats.effect.{ContextShift, IO}
 import doobie._
 import doobie.implicits._
@@ -30,7 +31,7 @@ trait ExtractorFixture extends SandboxFixture with PostgresAround with Types {
     666, // doesn't matter, will/must be overriden in the test cases
     LedgerOffset(LedgerOffset.Value.Boundary(LedgerOffset.LedgerBoundary.LEDGER_BEGIN)),
     SnapshotEndSetting.Head,
-    "Bob",
+    OneAnd("Bob", List.empty),
     TlsConfiguration(
       enabled = false,
       None,
@@ -39,6 +40,8 @@ trait ExtractorFixture extends SandboxFixture with PostgresAround with Types {
     )
   )
 
+  protected def configureExtractor(ec: ExtractorConfig): ExtractorConfig = ec
+
   protected lazy val target: PostgreSQLTarget = PostgreSQLTarget(
     connectUrl = postgresFixture.jdbcUrl,
     user = "test",
@@ -87,7 +90,7 @@ trait ExtractorFixture extends SandboxFixture with PostgresAround with Types {
   protected var extractor: Extractor[PostgreSQLTarget] = _
 
   protected def run(): Unit = {
-    val config: ExtractorConfig = baseConfig.copy(ledgerPort = getSandboxPort)
+    val config: ExtractorConfig = configureExtractor(baseConfig.copy(ledgerPort = getSandboxPort))
 
     extractor = new Extractor(config, target)
 

extractor/src/test/resources/damls/RecordsAndVariants.daml

@@ -51,4 +51,34 @@ suite =
           foo = [Some [1, 2, 3], None, Some [4, 5, 6], None, Some [7, 8, 9]]
           baz = RecordAB with
             foo = "foo"
-            baz = False
+            baz = False
+
+template Solo
+  with
+    party: Party
+    tick: Int
+  where
+    signatory party
+
+template Duo
+  with
+    party: Party
+    other: Party
+    tick: Int
+  where
+    signatory party
+    observer other
+
+multiParty = scenario do
+  bob <- getParty "Bob"
+  alice <- getParty "Alice"
+  quux <- getParty "Quux"
+  -- visible to Bob,Alice: 1, 2, 4, 5, 7
+  submit bob do create (Solo bob 1)
+  submit alice do create (Solo alice 2)
+  submit quux do create (Solo quux 3)
+  submit bob do create (Duo bob quux 4)
+  submit quux do
+    create (Duo quux alice 5)
+    create (Duo quux quux 6)
+  submit bob do create (Duo bob alice 7) -- ensure not seen twice

extractor/src/test/suite/scala/com/digitalasset/extractor/MultiPartySpec.scala

@@ -0,0 +1,53 @@
+// Copyright (c) 2019 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.digitalasset.extractor
+
+import com.digitalasset.extractor.config.ExtractorConfig
+import com.digitalasset.extractor.services.{CustomMatchers, ExtractorFixtureAroundAll}
+import com.digitalasset.ledger.api.testing.utils.SuiteResourceManagementAroundAll
+import com.digitalasset.platform.sandbox.persistence.PostgresAroundAll
+
+import org.scalatest._
+import java.io.File
+
+import scalaz._
+import scalaz.std.list._
+import scalaz.std.option._
+import scalaz.syntax.foldable._
+
+@SuppressWarnings(Array("org.wartremover.warts.Any"))
+class MultiPartySpec
+    extends FlatSpec
+    with Suite
+    with PostgresAroundAll
+    with SuiteResourceManagementAroundAll
+    with ExtractorFixtureAroundAll
+    with Inside
+    with Matchers
+    with CustomMatchers {
+
+  override protected def darFile = new File("extractor/RecordsAndVariants.dar")
+
+  override def scenario: Option[String] = Some("RecordsAndVariants:multiParty")
+
+  override def configureExtractor(ec: ExtractorConfig): ExtractorConfig = {
+    val ec2 = super.configureExtractor(ec)
+    ec2.copy(parties = OneAnd("Alice", ec2.parties.toList))
+  }
+
+  "Contracts" should "contain the visible contracts" in {
+    val ticks = getContracts.map { ct =>
+      for {
+        o <- ct.create_arguments.asObject
+        tick <- o("tick")
+        n <- tick.asNumber
+        i <- n.toInt
+      } yield i
+    }
+
+    val expected = List(1, 2, 4, 5, 7).map(some)
+
+    ticks should contain theSameElementsAs expected
+  }
+}