Typed ACS and transaction streams for Java codegen

docs/source/app-dev/bindings-java/quickstart/template-root/pom.xml

@@ -9,8 +9,8 @@
 
     <properties>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <maven.compiler.source>1.8</maven.compiler.source>
-        <maven.compiler.target>1.8</maven.compiler.target>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
         <daml-codegen-java.output>${project.build.directory}/generated-sources/iou</daml-codegen-java.output>
         <ledgerhost>localhost</ledgerhost>
         <ledgerport>6865</ledgerport>

docs/source/app-dev/bindings-java/quickstart/template-root/src/main/java/com/daml/quickstart/iou/IouMain.java

@@ -4,6 +4,7 @@
 package com.daml.quickstart.iou;
 
 import com.daml.ledger.javaapi.data.*;
+import com.daml.ledger.rxjava.ContractUtil;
 import com.daml.ledger.rxjava.DamlLedgerClient;
 import com.daml.ledger.rxjava.LedgerClient;
 import com.daml.quickstart.model.iou.Iou;
@@ -59,14 +60,14 @@ public static void main(String[] args) {
 
     client
         .getActiveContractSetClient()
-        .getActiveContracts(iouFilter, true)
+        .getActiveContracts(ContractUtil.of(Iou.COMPANION), Collections.singleton(party), true)
         .blockingForEach(
-            response -> {
-              response
+            activeContracts -> {
+              activeContracts
                   .getOffset()
                   .ifPresent(offset -> acsOffset.set(new LedgerOffset.Absolute(offset)));
-              response.getCreatedEvents().stream()
-                  .map(Iou.Contract::fromCreatedEvent)
+              activeContracts
+                  .getContracts()
                   .forEach(
                       contract -> {
                         long id = idCounter.getAndIncrement();

language-support/java/bindings-rxjava/src/main/java/com/daml/ledger/rxjava/ActiveContractsClient.java

@@ -3,9 +3,11 @@
 
 package com.daml.ledger.rxjava;
 
+import com.daml.ledger.javaapi.data.ActiveContracts;
 import com.daml.ledger.javaapi.data.GetActiveContractsResponse;
 import com.daml.ledger.javaapi.data.TransactionFilter;
 import io.reactivex.Flowable;
+import java.util.Set;
 
 /** An RxJava version of {@link com.daml.ledger.api.v1.ActiveContractsServiceGrpc} */
 public interface ActiveContractsClient {
@@ -15,4 +17,31 @@ Flowable<GetActiveContractsResponse> getActiveContracts(
 
   Flowable<GetActiveContractsResponse> getActiveContracts(
       TransactionFilter filter, boolean verbose, String accessToken);
+
+  /**
+   * Get active Contracts
+   *
+   * @param contractUtil Utilities for specified type of contract. It can be instantiated with
+   *     <code>ContractTypeCompanion</code>
+   * @param parties Set of parties to be included in the transaction filter.
+   * @param verbose If enabled, values served over the API will contain more information than
+   *     strictly necessary to interpret the data.
+   * @return Flowable of active contracts of type <code>Ct</code>
+   */
+  <Ct> Flowable<ActiveContracts<Ct>> getActiveContracts(
+      ContractUtil<Ct> contractUtil, Set<String> parties, boolean verbose);
+
+  /**
+   * Get active Contracts
+   *
+   * @param contractUtil Utilities for specified type of contract. It can be instantiated with
+   *     <code>ContractTypeCompanion</code>
+   * @param parties Set of parties to be included in the transaction filter.
+   * @param verbose If enabled, values served over the API will contain more information than
+   *     strictly necessary to interpret the data.
+   * @param accessToken Access token for authentication.
+   * @return Active contracts of type <code>Ct</code>
+   */
+  <Ct> Flowable<ActiveContracts<Ct>> getActiveContracts(
+      ContractUtil<Ct> contractUtil, Set<String> parties, boolean verbose, String accessToken);
 }

language-support/java/bindings-rxjava/src/main/java/com/daml/ledger/rxjava/ContractUtil.java

@@ -0,0 +1,64 @@
+// Copyright (c) 2022 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.rxjava;
+
+import com.daml.ledger.javaapi.data.*;
+import com.daml.ledger.javaapi.data.codegen.Contract;
+import com.daml.ledger.javaapi.data.codegen.ContractCompanion;
+import com.daml.ledger.javaapi.data.codegen.InterfaceCompanion;
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+/**
+ * This class contains utilities to decode a <code>CreatedEvent</code> and create a <code>
+ * TransactionFilter</code> by provider parties It can only be instantiated with a subtype of <code>
+ * ContractCompanion</code>
+ */
+public final class ContractUtil<Ct> {
+  private final FromCreatedEventFunc<CreatedEvent, Ct> fromCreatedEvent;
+
+  private final Filter filter;
+
+  private ContractUtil(FromCreatedEventFunc<CreatedEvent, Ct> fromCreatedEvent, Filter filter) {
+    this.fromCreatedEvent = fromCreatedEvent;
+    this.filter = filter;
+  }
+
+  public static <Ct> ContractUtil<Ct> of(ContractCompanion<Ct, ?, ?> companion) {
+    Filter filter =
+        new InclusiveFilter(Collections.singleton(companion.TEMPLATE_ID), Collections.emptyMap());
+    return new ContractUtil<>(companion::fromCreatedEvent, filter);
+  }
+
+  public static <Cid, View> ContractUtil<Contract<Cid, View>> of(
+      InterfaceCompanion<?, Cid, View> companion) {
+    Filter filter =
+        new InclusiveFilter(
+            Collections.emptySet(),
+            Collections.singletonMap(companion.TEMPLATE_ID, Filter.Interface.INCLUDE_VIEW));
+    return new ContractUtil<>(companion::fromCreatedEvent, filter);
+  }
+
+  public Ct toContract(CreatedEvent createdEvent) throws IllegalArgumentException {
+    return fromCreatedEvent.apply(createdEvent);
+  }
+
+  public TransactionFilter transactionFilter(Set<String> parties) {
+    return transactionFilter(filter, parties);
+  }
+
+  private static TransactionFilter transactionFilter(Filter filter, Set<String> parties) {
+    Map<String, Filter> partyToFilters =
+        parties.stream().collect(Collectors.toMap(Function.identity(), x -> filter));
+    return new FiltersByParty(partyToFilters);
+  }
+
+  @FunctionalInterface
+  private interface FromCreatedEventFunc<T, R> {
+    R apply(T t) throws IllegalArgumentException;
+  }
+}

language-support/java/bindings-rxjava/src/main/java/com/daml/ledger/rxjava/TransactionsClient.java

@@ -3,10 +3,7 @@
 
 package com.daml.ledger.rxjava;
 
-import com.daml.ledger.javaapi.data.LedgerOffset;
-import com.daml.ledger.javaapi.data.Transaction;
-import com.daml.ledger.javaapi.data.TransactionFilter;
-import com.daml.ledger.javaapi.data.TransactionTree;
+import com.daml.ledger.javaapi.data.*;
 import io.reactivex.Flowable;
 import io.reactivex.Single;
 import java.util.Set;
@@ -30,6 +27,39 @@ Flowable<Transaction> getTransactions(
   Flowable<Transaction> getTransactions(
       LedgerOffset begin, TransactionFilter filter, boolean verbose, String accessToken);
 
+  /**
+   * Get contracts
+   *
+   * @param contractUtil Utilities for specified type of contract. It can be instantiated with
+   *     <code>ContractTypeCompanion</code>
+   * @param begin begin offset.
+   * @param parties Set of parties to be included in the transaction filter.
+   * @param verbose If enabled, values served over the API will contain more information than
+   *     strictly necessary to interpret the data.
+   * @return Flowable of contract type <code>Ct</code>
+   */
+  <Ct> Flowable<Ct> getContracts(
+      ContractUtil<Ct> contractUtil, LedgerOffset begin, Set<String> parties, boolean verbose);
+
+  /**
+   * Get contracts
+   *
+   * @param contractUtil Utilities for specified type of contract. It can be instantiated with
+   *     <code>ContractTypeCompanion</code>
+   * @param begin begin offset.
+   * @param parties Set of parties to be included in the transaction filter.
+   * @param verbose If enabled, values served over the API will contain more information than
+   *     strictly necessary to interpret the data.
+   * @param accessToken Access token for authentication.
+   * @return Flowable of contract type <code>Ct</code>
+   */
+  <Ct> Flowable<Ct> getContracts(
+      ContractUtil<Ct> contractUtil,
+      LedgerOffset begin,
+      Set<String> parties,
+      boolean verbose,
+      String accessToken);
+
   Flowable<TransactionTree> getTransactionsTrees(
       LedgerOffset begin, LedgerOffset end, TransactionFilter filter, boolean verbose);
 

language-support/java/bindings-rxjava/src/main/java/com/daml/ledger/rxjava/grpc/ActiveContractClientImpl.java

@@ -6,15 +6,17 @@
 import com.daml.grpc.adapter.ExecutionSequencerFactory;
 import com.daml.ledger.api.v1.ActiveContractsServiceGrpc;
 import com.daml.ledger.api.v1.ActiveContractsServiceOuterClass;
-import com.daml.ledger.javaapi.data.GetActiveContractsRequest;
-import com.daml.ledger.javaapi.data.GetActiveContractsResponse;
-import com.daml.ledger.javaapi.data.TransactionFilter;
+import com.daml.ledger.javaapi.data.*;
 import com.daml.ledger.rxjava.ActiveContractsClient;
+import com.daml.ledger.rxjava.ContractUtil;
 import com.daml.ledger.rxjava.grpc.helpers.StubHelper;
 import com.daml.ledger.rxjava.util.ClientPublisherFlowable;
 import io.grpc.Channel;
 import io.reactivex.Flowable;
+import java.util.List;
 import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
 import org.checkerframework.checker.nullness.qual.NonNull;
 
 public class ActiveContractClientImpl implements ActiveContractsClient {
@@ -56,4 +58,36 @@ public Flowable<GetActiveContractsResponse> getActiveContracts(
       @NonNull TransactionFilter filter, boolean verbose, @NonNull String accessToken) {
     return getActiveContracts(filter, verbose, Optional.of(accessToken));
   }
+
+  private <Ct> Flowable<ActiveContracts<Ct>> getActiveContracts(
+      ContractUtil<Ct> contractUtil,
+      Set<String> parties,
+      boolean verbose,
+      Optional<String> accessToken) {
+    TransactionFilter filter = contractUtil.transactionFilter(parties);
+
+    Flowable<GetActiveContractsResponse> responses =
+        getActiveContracts(filter, verbose, accessToken);
+    return responses.map(
+        response -> {
+          List<Ct> activeContracts =
+              response.getCreatedEvents().stream()
+                  .map(contractUtil::toContract)
+                  .collect(Collectors.toList());
+          return new ActiveContracts<>(
+              response.getOffset().orElse(""), activeContracts, response.getWorkflowId());
+        });
+  }
+
+  @Override
+  public <Ct> Flowable<ActiveContracts<Ct>> getActiveContracts(
+      ContractUtil<Ct> contractUtil, Set<String> parties, boolean verbose) {
+    return getActiveContracts(contractUtil, parties, verbose, Optional.empty());
+  }
+
+  @Override
+  public <Ct> Flowable<ActiveContracts<Ct>> getActiveContracts(
+      ContractUtil<Ct> contractUtil, Set<String> parties, boolean verbose, String accessToken) {
+    return getActiveContracts(contractUtil, parties, verbose, Optional.of(accessToken));
+  }
 }

language-support/java/bindings-rxjava/src/main/java/com/daml/ledger/rxjava/grpc/TransactionClientImpl.java

@@ -7,6 +7,7 @@
 import com.daml.ledger.api.v1.TransactionServiceGrpc;
 import com.daml.ledger.api.v1.TransactionServiceOuterClass;
 import com.daml.ledger.javaapi.data.*;
+import com.daml.ledger.rxjava.ContractUtil;
 import com.daml.ledger.rxjava.TransactionsClient;
 import com.daml.ledger.rxjava.grpc.helpers.StubHelper;
 import com.daml.ledger.rxjava.util.ClientPublisherFlowable;
@@ -16,6 +17,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.Future;
+import java.util.stream.Collectors;
 
 public final class TransactionClientImpl implements TransactionsClient {
   private final String ledgerId;
@@ -73,6 +75,40 @@ public Flowable<Transaction> getTransactions(
     return getTransactions(begin, end, filter, verbose, Optional.of(accessToken));
   }
 
+  private <Ct> Flowable<Ct> getContracts(
+      ContractUtil<Ct> contractUtil,
+      LedgerOffset begin,
+      Set<String> parties,
+      boolean verbose,
+      Optional<String> accessToken) {
+    TransactionFilter filter = contractUtil.transactionFilter(parties);
+    Flowable<Transaction> transactions = getTransactions(begin, filter, verbose, accessToken);
+    Flowable<CreatedEvent> createdEvents =
+        transactions.concatMapIterable(
+            tx ->
+                tx.getEvents().stream()
+                    .filter(e -> e instanceof CreatedEvent)
+                    .map(e -> (CreatedEvent) e)
+                    .collect(Collectors.toList()));
+    return createdEvents.map(contractUtil::toContract);
+  }
+
+  @Override
+  public <Ct> Flowable<Ct> getContracts(
+      ContractUtil<Ct> contractUtil, LedgerOffset begin, Set<String> parties, boolean verbose) {
+    return getContracts(contractUtil, begin, parties, verbose, Optional.empty());
+  }
+
+  @Override
+  public <Ct> Flowable<Ct> getContracts(
+      ContractUtil<Ct> contractUtil,
+      LedgerOffset begin,
+      Set<String> parties,
+      boolean verbose,
+      String accessToken) {
+    return getContracts(contractUtil, begin, parties, verbose, Optional.of(accessToken));
+  }
+
   private Flowable<Transaction> getTransactions(
       LedgerOffset begin, TransactionFilter filter, boolean verbose, Optional<String> accessToken) {
     TransactionServiceOuterClass.GetTransactionsRequest request =

language-support/java/bindings/src/main/java/com/daml/ledger/javaapi/data/ActiveContracts.java

@@ -0,0 +1,69 @@
+// Copyright (c) 2022 Digital Asset (Switzerland) GmbH and/or its affiliates. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+package com.daml.ledger.javaapi.data;
+
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+public final class ActiveContracts<Ct> implements WorkflowEvent {
+
+  private final String offset;
+
+  private final List<Ct> activeContracts;
+
+  private final String workflowId;
+
+  public ActiveContracts(
+      @NonNull String offset, @NonNull List<Ct> activeContracts, String workflowId) {
+    this.offset = offset;
+    this.activeContracts = activeContracts;
+    this.workflowId = workflowId;
+  }
+
+  @NonNull
+  public Optional<String> getOffset() {
+    // Empty string indicates that the field is not present in the protobuf.
+    return Optional.of(offset).filter(off -> !offset.equals(""));
+  }
+
+  @NonNull
+  public List<@NonNull Ct> getContracts() {
+    return activeContracts;
+  }
+
+  @NonNull
+  public String getWorkflowId() {
+    return workflowId;
+  }
+
+  @Override
+  public String toString() {
+    return "ActiveContracts{"
+        + "offset='"
+        + offset
+        + '\''
+        + ", activeContracts="
+        + activeContracts
+        + ", workflowId="
+        + workflowId
+        + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    ActiveContracts<?> that = (ActiveContracts<?>) o;
+    return Objects.equals(offset, that.offset)
+        && Objects.equals(activeContracts, that.activeContracts)
+        && Objects.equals(workflowId, that.workflowId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(offset, activeContracts, workflowId);
+  }
+}