Handle error acks

client/src/main/java/io/spine/client/Client.java

@@ -22,6 +22,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.ImmutableList;
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.inprocess.InProcessChannelBuilder;
@@ -30,6 +31,7 @@
 import io.spine.client.grpc.CommandServiceGrpc.CommandServiceBlockingStub;
 import io.spine.client.grpc.QueryServiceGrpc;
 import io.spine.client.grpc.QueryServiceGrpc.QueryServiceBlockingStub;
+import io.spine.core.Ack;
 import io.spine.core.Command;
 import io.spine.core.TenantId;
 import io.spine.core.UserId;
@@ -102,6 +104,16 @@ public class Client implements AutoCloseable {
     /** Active subscriptions maintained by the client. */
     private final Subscriptions subscriptions;
 
+    /**
+     * The handler for errors that may occur during asynchronous requests initiated by this client.
+     */
+    private final @Nullable ErrorHandler streamingErrorHandler;
+
+    /**
+     * The handler for errors returned from server side in response to posted messages.
+     */
+    private final @Nullable ServerErrorHandler serverErrorHandler;
+
     /**
      * Creates a builder for a client connected to the specified address.
      *
@@ -156,7 +168,9 @@ private Client(Builder builder) {
         this.shutdownTimeout = checkNotNull(builder.shutdownTimeout);
         this.commandService = CommandServiceGrpc.newBlockingStub(channel);
         this.queryService = QueryServiceGrpc.newBlockingStub(channel);
-        this.subscriptions = new Subscriptions(channel);
+        this.streamingErrorHandler = builder.streamingErrorHandler;
+        this.serverErrorHandler = builder.serverErrorHandler;
+        this.subscriptions = new Subscriptions(channel, streamingErrorHandler, serverErrorHandler);
     }
 
     /**
@@ -214,7 +228,14 @@ public boolean isOpen() {
      */
     public ClientRequest onBehalfOf(UserId user) {
         checkNotDefaultArg(user);
-        return new ClientRequest(user, this);
+        ClientRequest request = new ClientRequest(user, this);
+        if (streamingErrorHandler != null) {
+            request.onStreamingError(streamingErrorHandler);
+        }
+        if (serverErrorHandler != null) {
+            request.onServerError(serverErrorHandler);
+        }
+        return request;
     }
 
     /**
@@ -269,8 +290,9 @@ ActorRequestFactory requestOf(UserId user) {
     /**
      * Posts the command to the {@code CommandService}.
      */
-    void post(Command c) {
-        commandService.post(c);
+    Ack post(Command c) {
+        Ack ack = commandService.post(c);
+        return ack;
     }
 
     /**
@@ -324,6 +346,9 @@ public static final class Builder {
         /** The ID of the user for performing requests on behalf of a non-logged in user. */
         private UserId guestUser = DEFAULT_GUEST_ID;
 
+        private @Nullable ErrorHandler streamingErrorHandler;
+        private @Nullable ServerErrorHandler serverErrorHandler;
+
         private Builder(ManagedChannel channel) {
             this.channel = checkNotNull(channel);
         }
@@ -404,6 +429,27 @@ public Builder shutdownTimout(long timeout, TimeUnit timeUnit) {
             return this;
         }
 
+        /**
+         * Assigns a default handler for streaming errors for the asynchronous requests
+         * initiated by the client.
+         */
+        @CanIgnoreReturnValue
+        public Builder onStreamingError(ErrorHandler handler) {
+            this.streamingErrorHandler = checkNotNull(handler);
+            return this;
+        }
+
+        /**
+         * Assigns a default handler for an error occurred on the server-side (such as
+         * validation error) in response to a message posted by the client.
+         */
+        @CanIgnoreReturnValue
+        public Builder onServerError(ServerErrorHandler handler) {
+            checkNotNull(handler);
+            this.serverErrorHandler = handler;
+            return this;
+        }
+
         /**
          * Creates a new instance of the client.
          */

client/src/main/java/io/spine/client/ClientRequest.java

@@ -20,10 +20,13 @@
 
 package io.spine.client;
 
+import com.google.errorprone.annotations.CanIgnoreReturnValue;
+import com.google.errorprone.annotations.OverridingMethodsMustInvokeSuper;
 import io.spine.base.CommandMessage;
 import io.spine.base.EntityState;
 import io.spine.base.EventMessage;
 import io.spine.core.UserId;
+import org.checkerframework.checker.nullness.qual.Nullable;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.spine.util.Preconditions2.checkNotDefaultArg;
@@ -47,6 +50,9 @@ public class ClientRequest {
     private final UserId user;
     private final Client client;
 
+    private @Nullable ErrorHandler streamingErrorHandler;
+    private @Nullable ServerErrorHandler serverErrorHandler;
+
     ClientRequest(UserId user, Client client) {
         checkNotDefaultArg(user);
         this.user = user;
@@ -55,12 +61,15 @@ public class ClientRequest {
 
     ClientRequest(ClientRequest parent) {
         this(parent.user, parent.client);
+        this.streamingErrorHandler = parent.streamingErrorHandler;
+        this.serverErrorHandler = parent.serverErrorHandler;
     }
 
     /**
      * Creates a builder for customizing command request.
      */
     public CommandRequest command(CommandMessage c) {
+        checkNotNull(c);
         return new CommandRequest(this, c);
     }
 
@@ -100,4 +109,36 @@ protected final UserId user() {
     protected final Client client() {
         return client;
     }
+
+    /**
+     * Assigns a handler for errors occurred when delivering messages from the server.
+     *
+     * <p>If such an error occurs, no more results are expected from the server.
+     */
+    @CanIgnoreReturnValue
+    @OverridingMethodsMustInvokeSuper
+    public ClientRequest onStreamingError(ErrorHandler handler) {
+        this.streamingErrorHandler = checkNotNull(handler);
+        return this;
+    }
+
+    /**
+     * Assigns a handler for an error occurred on the server-side (such as validation error)
+     * in response to posting a request.
+     */
+    @CanIgnoreReturnValue
+    @OverridingMethodsMustInvokeSuper
+    public ClientRequest onServerError(ServerErrorHandler handler) {
+        checkNotNull(handler);
+        this.serverErrorHandler = handler;
+        return this;
+    }
+
+    final @Nullable ErrorHandler streamingErrorHandler() {
+        return streamingErrorHandler;
+    }
+
+    final @Nullable ServerErrorHandler serverErrorHandler() {
+        return serverErrorHandler;
+    }
 }