Command Completion Service Sample

Is there sample on how to use command completion service? Esp how to use Checkpoint? Thanks. My understanding is that the command completion service is the one that will provide definite answer on the command execution result and I would like to try it out.

I don’t think we have any sample using the command completion service. Please note that checkpoints are there for backwards compatibility but they are no longer used, as the new time model and command deduplication mechanism obsoleted them since 1.0. What you can do by using the command submission and completions services is have a fully event-driven architecture where you track command results on the client. This buys you some overall scalability compared to using the command service where command tracking is performed on the Ledger API server (where you share resources with other clients).

@stefanobaghino-da

I am using the class CommandCompletionClientImpl in package com.daml.ledger.rxjava.grpc to build a command submission client. Is it the correct way to use it?

    NettyChannelBuilder channelBuilder = NettyChannelBuilder.forAddress(ledgerhost, ledgerport);
    NettyChannelBuilder channelBuilder = NettyChannelBuilder
                                            .forAddress(ledgerhost, ledgerport)
                                            .usePlaintext();
    ManagedChannel channel = channelBuilder.build();
    
    CommandSubmissionClientImpl submissionClient = new CommandSubmissionClientImpl(LEDGER_ID, channel, Optional.empty());
    CommandCompletionClientImpl completionClient = new CommandCompletionClientImpl(LEDGER_ID, channel, new SingleThreadExecutionSequencerPool("test"), Optional.empty());

The CommandCompletionClientImpl requires a ExecutionSequencerFactory in the constructor. Is there any special requirement for that? Can I just create one from SingleThreadExecutionSequencerPool? How does it get used?

Now I’m getting this error

01:32:30.759 [client-1] ERROR c.d.g.a.SingleThreadExecutionSequencer [  : -  ] - Unhandled exception in SingleThreadExecutionSequencer.
io.reactivex.exceptions.OnErrorNotImplementedException: INVALID_ARGUMENT: Missing field: begin
at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
at io.reactivex.internal.subscribers.LambdaSubscriber.onError(LambdaSubscriber.java:79)
at io.reactivex.internal.operators.flowable.FlowableFlattenIterable$FlattenIterableSubscriber.checkTerminated(FlowableFlattenIterable.java:396)
at io.reactivex.internal.operators.flowable.FlowableFlattenIterable$FlattenIterableSubscriber.drain(FlowableFlattenIterable.java:256)
at io.reactivex.internal.operators.flowable.FlowableFlattenIterable$FlattenIterableSubscriber.onError(FlowableFlattenIterable.java:182)
at io.reactivex.internal.subscribers.BasicFuseableSubscriber.onError(BasicFuseableSubscriber.java:101)
at com.daml.grpc.adapter.client.rs.BufferingResponseObserver.lambda$onError$3(BufferingResponseObserver.java:81)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)
Caused by: io.grpc.StatusRuntimeException: INVALID_ARGUMENT: Missing field: begin
at io.grpc.Status.asRuntimeException(Status.java:533)
at io.grpc.stub.ClientCalls$StreamObserverToCallListenerAdapter.onClose(ClientCalls.java:453)
at io.grpc.internal.ClientCallImpl.closeObserver(ClientCallImpl.java:426)
at io.grpc.internal.ClientCallImpl.access$500(ClientCallImpl.java:66)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.close(ClientCallImpl.java:689)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl.access$900(ClientCallImpl.java:577)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInternal(ClientCallImpl.java:751)
at io.grpc.internal.ClientCallImpl$ClientStreamListenerImpl$1StreamClosed.runInContext(ClientCallImpl.java:740)
at io.grpc.internal.ContextRunnable.run(ContextRunnable.java:37)
at io.grpc.internal.SerializingExecutor.run(SerializingExecutor.java:123)
... 3 common frames omitted

Found out where that error came from. It happens because I got another DamlLedgerClient.connect() running before channel.build(). It now works fine after I delete it.

I would recommend not instantiating the individual clients by hand. Those clients are best used via DamlLedgerClient, that you’ve later shown to already use. Just access commandSubmissionClient and commandCompletionClient on your DamlLedgerClient object to access ready-built clients for both services.