Merge pull request #151 from SpineEventEngine/observer-on-error-for-integr-events-posting

Use `StreamObserver.onError()` while posting integration events

Author: AlexanderLitus

client/src/main/java/org/spine3/SPI.java

@@ -31,7 +31,7 @@
  *
  * <p>SPI is used to enable framework extension and replaceable components (implement a new storage, etc).
  *
- * <p>See Effective Java 2nd Edition, chapter 2, item 1 for more info about service provider framework pattern.
+ * <p>See "Effective Java 2nd Edition", chapter 2, item 1 for more info about service provider framework pattern.
  *
  * @author Alexander Litus
  */

client/src/main/java/org/spine3/base/Responses.java

@@ -58,39 +58,4 @@ public static boolean isOk(Response response) {
         final boolean result = response.getStatusCase() == OK;
         return result;
     }
-
-    /**
-     * Checks if the response is `unsupported event`.
-     *
-     * @return {@code true} if the passed response represents `unsupported event` error,
-     *         {@code false} otherwise
-     */
-    public static boolean isUnsupportedEvent(Response response) {
-        if (isError(response)) {
-            final Error error = response.getError();
-            final boolean isUnsupported = error.getCode() == EventValidationError.UNSUPPORTED_EVENT.getNumber();
-            return isUnsupported;
-        }
-        return false;
-    }
-
-    /**
-     * Checks if the response is `invalid message`.
-     *
-     * @return {@code true} if the passed response represents `invalid message` error,
-     *         {@code false} otherwise
-     */
-    public static boolean isInvalidMessage(Response response) {
-        if (isError(response)) {
-            final ValidationError error = response.getError().getValidationError();
-            final boolean isInvalid = !error.getConstraintViolationList().isEmpty();
-            return isInvalid;
-        }
-        return false;
-    }
-
-    private static boolean isError(Response response) {
-        final boolean isError = response.getStatusCase() == ERROR;
-        return isError;
-    }
 }

client/src/main/proto/spine/annotations.proto

@@ -0,0 +1,107 @@
+//
+// Copyright 2016, TeamDev Ltd. All rights reserved.
+//
+// Redistribution and use in source and/or binary forms, with or without
+// modification, must retain the above copyright notice and the following
+// disclaimer.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+//
+syntax = "proto3";
+
+// We do not define the package for this file to allow shorter options for user-defined types.
+// This would allow to write:
+//
+//     option (internal) = true;
+//
+// instead of:
+//
+//      option (spine.base.internal) = true;
+//
+
+option java_generate_equals_and_hash = false;
+option java_multiple_files = true;
+option java_outer_classname = "AnnotationsProto";
+option java_package = "org.spine3.annotations";
+
+import "google/protobuf/descriptor.proto";
+
+// TODO:2016-02-18:alexander.litus: obtain globally unique field numbers for opts from Google.
+
+
+extend google.protobuf.FieldOptions {
+    // Indicates a field which is internal to Spine, not part of the public API, and should not be used by users of the framework.
+    //
+    // If you plan to implement an extension of the framework, which is going to be
+    // wired into the framework, you may use the internal parts. Please consult with the Spine
+    // team, as the internal APIs do not have the same stability API guarantee as public ones.
+    //
+    // See `SPI` option if you plan to write an extension of the framework.
+    bool internal = 58000;
+
+    // Indicates a file which contains elements of Service Provider Interface (SPI).
+    // SPI is used to enable framework extension and replaceable components (implement a new storage, etc).
+    // See "Effective Java 2nd Edition", chapter 2, item 1 for more info about service provider framework pattern.
+    bool SPI = 58001;
+
+    // Indicates a field that can change at any time, and has no guarantee of API stability and backward-compatibility.
+    //
+    // Usage guidelines:
+    // 1. This annotation is used only on public API. Internal interfaces should not use it.
+    // 2. This annotation can only be added to new API. Adding it to an existing API is considered API-breaking.
+    // 3. Removing this annotation from an API gives it stable status.
+    bool experimental = 58002;
+
+    // Signifies that a public API is subject to incompatible changes, or even removal, in a future release.
+    // An API bearing this annotation is exempt from any compatibility guarantees made by its containing library.
+    // Note that the presence of this annotation implies nothing about the quality of the API in question,
+    // only the fact that it is not "API-frozen."
+    // It is generally safe for applications to depend on beta APIs, at the cost of some extra work during upgrades.
+    bool beta = 58003;
+}
+
+extend google.protobuf.MessageOptions {
+    // Indicates a file which is internal to Spine, not part of the public API, and should not be used by users of the framework.
+    // See more info in `annotations.proto`.
+    bool internal_type = 58004;
+
+    // Indicates a file which contains elements of Service Provider Interface (SPI).
+    // See more info in `annotations.proto`.
+    bool SPI_type = 58005;
+
+    // Indicates a public API that can change at any time, and has no guarantee of API stability and backward-compatibility.
+    // See more info in `annotations.proto`.
+    bool experimental_type = 58006;
+
+    // Signifies that a public API is subject to incompatible changes, or even removal, in a future release.
+    // See more info in `annotations.proto`.
+    bool beta_type = 58007;
+}
+
+extend google.protobuf.FileOptions {
+    // Indicates a file which should not be used by users of the framework.
+    // See more info in `annotations.proto`.
+    bool internal_all = 58008;
+
+    // Indicates a file which contains elements of Service Provider Interface (SPI).
+    // See more info in `annotations.proto`.
+    bool SPI_all = 58009;
+
+    // Indicates a public API that can change at any time, and has no guarantee of API stability and backward-compatibility.
+    // See more info in `annotations.proto`.
+    bool experimental_all = 58010;
+
+    // Signifies that a public API is subject to incompatible changes, or even removal, in a future release.
+    // See more info in `annotations.proto`.
+    bool beta_all = 58011;
+}

client/src/main/proto/spine/base/command.proto

@@ -34,6 +34,8 @@ import "spine/users/user_id.proto";
 import "spine/users/tenant_id.proto";
 import "spine/time/zone_offset.proto";
 
+import "spine/annotations.proto";
+
 // Command identifier.
 message CommandId {
     string uuid = 1;
@@ -91,9 +93,8 @@ message CommandContext {
         // The recurrence pattern should be a separate type somewhere under `time` package.
 
         // The time when the command was scheduled.
-        // Is set automatically.
-        //TODO:2016-05-27:alexander.yevsyukov: Set automatically by whom? How do we use this attribute?
-        google.protobuf.Timestamp scheduling_time = 2;
+        // Is set automatically and used only by Spine when re-scheduling a command.
+        google.protobuf.Timestamp scheduling_time = 2 [(internal) = true];
     }
 
     // The schedule to execute the command.

client/src/test/java/org/spine3/base/ResponsesShould.java

@@ -21,26 +21,15 @@
 package org.spine3.base;
 
 import org.junit.Test;
-import org.spine3.validate.ConstraintViolation;
-
-import java.util.List;
 
 import static com.google.common.base.Preconditions.checkNotNull;
-import static com.google.common.collect.Lists.newArrayList;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.spine3.test.Tests.hasPrivateUtilityConstructor;
 
 @SuppressWarnings("InstanceMethodNamingConvention")
 public class ResponsesShould {
 
-    private static final Response RESPONSE_UNSUPPORTED_EVENT = Response.newBuilder()
-            .setError(Error.newBuilder()
-                           .setCode(EventValidationError.UNSUPPORTED_EVENT.getNumber()))
-            .build();
-
-    private static final Response RESPONSE_INVALID_MESSAGE = newInvalidMessageResponse();
-
     @Test
     public void have_private_constructor() {
         assertTrue(hasPrivateUtilityConstructor(Responses.class));
@@ -58,40 +47,9 @@ public void recognize_OK_response() {
 
     @Test
     public void return_false_if_not_OK_response() {
-        assertFalse(Responses.isOk(RESPONSE_INVALID_MESSAGE));
-    }
-
-    @Test
-    public void recognize_UNSUPPORTED_EVENT_response() {
-        assertTrue(Responses.isUnsupportedEvent(RESPONSE_UNSUPPORTED_EVENT));
-    }
-
-    @Test
-    public void return_false_if_not_UNSUPPORTED_EVENT_response() {
-        assertFalse(Responses.isUnsupportedEvent(Responses.ok()));
-    }
-
-    @Test
-    public void recognize_INVALID_MESSAGE_response() {
-        assertTrue(Responses.isInvalidMessage(RESPONSE_INVALID_MESSAGE));
-    }
-
-    @Test
-    public void return_false_if_not_INVALID_MESSAGES_response() {
-        assertFalse(Responses.isInvalidMessage(Responses.ok()));
-    }
-
-    private static Response newInvalidMessageResponse() {
-        final List<ConstraintViolation> violations = newArrayList(ConstraintViolation.getDefaultInstance());
-        final ValidationError validationError = ValidationError.newBuilder()
-                                                               .addAllConstraintViolation(violations)
-                                                               .build();
-        final Error error = Error.newBuilder()
-                                 .setValidationError(validationError)
-                                 .build();
-        final Response response = Response.newBuilder()
-                                          .setError(error)
-                                          .build();
-        return response;
+        final Response error = Response.newBuilder()
+                                       .setError(Error.getDefaultInstance())
+                                       .build();
+        assertFalse(Responses.isOk(error));
     }
 }

server/src/main/java/org/spine3/server/BoundedContext.java

@@ -26,10 +26,8 @@
 import io.grpc.stub.StreamObserver;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.spine3.base.Command;
 import org.spine3.base.Event;
 import org.spine3.base.EventContext;
-import org.spine3.base.Failure;
 import org.spine3.base.Response;
 import org.spine3.server.command.CommandBus;
 import org.spine3.server.command.CommandDispatcher;
@@ -52,7 +50,6 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
-import static org.spine3.base.Responses.isOk;
 import static org.spine3.protobuf.Messages.fromAny;
 import static org.spine3.protobuf.Messages.toAny;
 import static org.spine3.protobuf.Values.newStringValue;
@@ -195,36 +192,14 @@ private void checkStorageAssigned(Repository repository) {
 
     @Override
     public void notify(IntegrationEvent integrationEvent, StreamObserver<Response> responseObserver) {
-        /**
-         * TODO:2016-05-11:alexander.litus: use {@link StreamObserver#onError}
-         * instead of returning responses, see {@link CommandBus#post(Command, StreamObserver)}.
-         */
-        try {
-            final Message message = fromAny(integrationEvent.getMessage());
-            final Response response = validateIntegrationEventMessage(message);
-            responseObserver.onNext(response);
-            responseObserver.onCompleted();
-            if (isOk(response)) {
-                final Event event = toEvent(integrationEvent);
-                eventBus.post(event);
-            }
-        } catch (RuntimeException e) {
-            responseObserver.onError(e);
+        final Message eventMsg = fromAny(integrationEvent.getMessage());
+        final boolean isValid = eventBus.validate(eventMsg, responseObserver);
+        if (isValid) {
+            final Event event = toEvent(integrationEvent);
+            eventBus.post(event);
         }
     }
 
-    /**
-     * Validates an incoming integration event message.
-     *
-     * @param eventMessage a message to validate
-     * @return a response with {@code OK} value if the command is valid, or
-     *          with {@link org.spine3.base.Error}/{@link Failure} value otherwise
-     */
-    protected Response validateIntegrationEventMessage(Message eventMessage) {
-        final Response response = eventBus.validate(eventMessage);
-        return response;
-    }
-
     private static Event toEvent(IntegrationEvent integrationEvent) {
         final IntegrationEventContext sourceContext = integrationEvent.getContext();
         final StringValue producerId = newStringValue(sourceContext.getBoundedContextName());
@@ -346,6 +321,7 @@ public CommandBus getCommandBus() {
          *
          * @see #setEventStoreStreamExecutor(Executor)
          */
+        @SuppressWarnings("unused")
         public Builder setEventStore(EventStore eventStore) {
             checkState(eventStoreStreamExecutor == null, "eventStoreStreamExecutor already set.");
             this.eventStore = checkNotNull(eventStore);