Stand funnel tests

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

@@ -28,7 +28,6 @@ option java_outer_classname = "ClientProto";
 option java_package = "org.spine3.client";
 
 import "spine/annotations.proto";
-import "spine/base/event.proto";
 
 // Version of the code executed on the client.
 message CodeVersion {

client/src/main/proto/spine/client/client_service.proto

@@ -31,26 +31,11 @@ option java_generate_equals_and_hash = true;
 
 import "spine/annotations.proto";
 import "spine/base/command.proto";
-import "spine/base/event.proto";
 import "spine/base/response.proto";
 
-// A topic of interest the client can subscribe and unsubscribe.
-message Topic {
-    //TODO:2016-01-14:alexander.yevsyukov: Define this type. E.g. there can be some structure, which describes many
-    // points of interest at once. See Pub-sub for possible API inspiration. Chances are it's going to be one of underlying
-    // implementations.
-    string value = 1;
-}
 
 // A service for sending commands from clients.
 service ClientService {
     // Request to handle a command.
     rpc Post(base.Command) returns (base.Response);
-
-    // Request to receive events on the topic of interest.
-    rpc Subscribe(Topic) returns (stream base.Event);
-
-    // The request to unsubscribe from the topic.
-    // This should close the stream opened by `Subscribe` call with the same `Topic` value.
-    rpc Unsubscribe(Topic) returns (base.Response);
 }

client/src/main/proto/spine/client/entities.proto

@@ -28,11 +28,8 @@ option java_outer_classname = "EntitiesProto";
 option java_package = "org.spine3.client";
 
 import "google/protobuf/any.proto";
-import "google/protobuf/field_mask.proto";
 
 import "spine/annotations.proto";
-import "spine/ui/language.proto";
-import "spine/base/response.proto";
 
 // Represents an ID of an entity.
 //

client/src/main/proto/spine/client/query_service.proto

@@ -27,7 +27,6 @@ option java_multiple_files = true;
 option java_outer_classname = "QueryServiceProto";
 option java_generate_equals_and_hash = true;
 
-import "google/protobuf/any.proto";
 
 import "spine/annotations.proto";
 import "spine/client/query.proto";

client/src/main/proto/spine/client/subscription.proto

@@ -0,0 +1,83 @@
+//
+// 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";
+
+package spine.client;
+
+option (type_url_prefix) = "type.spine3.org";
+option java_generate_equals_and_hash = true;
+option java_multiple_files = true;
+option java_outer_classname = "SubscriptionProto";
+option java_package = "org.spine3.client";
+
+
+import "google/protobuf/any.proto";
+import "google/protobuf/field_mask.proto";
+
+import "spine/annotations.proto";
+import "spine/base/response.proto";
+import "spine/client/entities.proto";
+
+// An object defining a unit of subscription.
+//
+// Defines the target (entities and criteria) of subscription.
+message Topic {
+
+    // Defines the entity of interest, e.g. entity type URL and a set of subscription criteria.
+    Target target = 1;
+
+    // Field mask to be applied to the entity updates applicable to this topic.
+    //
+    // Applied to each of the entity state messages before returning in scope of SubscriptionUpdate.
+    google.protobuf.FieldMask field_mask = 2;
+
+    // Reserved for utility fields.
+    reserved 3 to 6;
+}
+
+// Wrapped set of read-side entity updates on a topic with the specific subscription ID.
+message SubscriptionUpdate {
+
+    // The ID of the current subscription.
+    SubscriptionId id = 1;
+
+    // Represents the base part of the response. I.e. whether the Topic subscription requires has been acked or not.
+    base.Response response = 2;
+
+    // Reserved for more subscription update attributes.
+    reserved 3 to 9;
+
+    // Entity updates packed as Any.
+    //
+    // Each of the update messages is affected by the field mask set for the current subscription.
+    repeated google.protobuf.Any updates = 10;
+}
+
+// The ID of the subscription.
+//
+// Created when the client subscribes to a topic.
+// See SubscriptionService#Subscribe(Topic).
+message SubscriptionId {
+
+    // Unique identifier of the subscription.
+    //
+    // Typically built using Java's UUID.toString() functionality.
+    string uuid = 1;
+}

client/src/main/proto/spine/client/subscription_service.proto

@@ -0,0 +1,46 @@
+//
+// 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";
+
+package spine.client;
+
+option (type_url_prefix) = "type.spine3.org";
+option java_package = "org.spine3.client.grpc";
+option java_multiple_files = true;
+option java_outer_classname = "SubscriptionServiceProto";
+option java_generate_equals_and_hash = true;
+
+
+import "spine/annotations.proto";
+import "spine/client/subscription.proto";
+import "spine/base/response.proto";
+
+// A service for subscribing to the read-side changes by clients.
+service SubscriptionService {
+
+    // Subscribe to a particular read-side updates.
+    //
+    // Topic defines the target of subscription and other attributes (like field masks).
+    // The result is a SubscriptionUpdate stream,
+    rpc Subscribe (Topic) returns (stream SubscriptionUpdate);
+
+    // Cancel the subscription by its ID.
+    rpc CancelSubscription (SubscriptionId) returns (base.Response);
+}

examples/src/main/java/org/spine3/examples/aggregate/ClientApp.java

@@ -31,7 +31,6 @@
 import org.spine3.base.Response;
 import org.spine3.client.CommandFactory;
 import org.spine3.client.grpc.ClientServiceGrpc;
-import org.spine3.client.grpc.Topic;
 import org.spine3.examples.aggregate.command.AddOrderLine;
 import org.spine3.examples.aggregate.command.CreateOrder;
 import org.spine3.examples.aggregate.command.PayForOrder;
@@ -66,9 +65,9 @@ public class ClientApp {
     private static final int SHUTDOWN_TIMEOUT_SEC = 5;
 
     private final CommandFactory commandFactory;
-    private final Topic topic = Topic.getDefaultInstance();
     private final ManagedChannel channel;
     private final ClientServiceGrpc.ClientServiceBlockingStub blockingClient;
+    // TODO[alex.tymchenko]: switch to SubscriptionService instead.  
     private final ClientServiceGrpc.ClientServiceStub nonBlockingClient;
 
     private final StreamObserver<Event> observer = new StreamObserver<Event>() {
@@ -140,14 +139,10 @@ private Command payForOrder(OrderId orderId) {
         return commandFactory.create(msg);
     }
 
-    private void subscribe() {
-        nonBlockingClient.subscribe(topic, observer);
-    }
 
     /** Sends requests to the server. */
     public static void main(String[] args) throws InterruptedException {
         final ClientApp client = new ClientApp(SERVICE_HOST, DEFAULT_CLIENT_SERVICE_PORT);
-        client.subscribe();
 
         final List<Command> requests = client.generateRequests();
 
@@ -178,7 +173,6 @@ private List<Command> generateRequests() {
      * @throws InterruptedException if waiting is interrupted.
      */
     private void shutdown() throws InterruptedException {
-        blockingClient.unsubscribe(topic);
         channel.shutdown().awaitTermination(SHUTDOWN_TIMEOUT_SEC, SECONDS);
     }
 

gradle/wrapper/gradle-wrapper.properties

@@ -3,4 +3,4 @@ distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-2.14.1-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-3.0-bin.zip

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

@@ -26,10 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.spine3.base.Command;
-import org.spine3.base.Event;
 import org.spine3.base.Response;
-import org.spine3.base.Responses;
-import org.spine3.client.grpc.Topic;
 import org.spine3.server.command.error.CommandException;
 import org.spine3.server.command.error.UnsupportedCommandException;
 import org.spine3.server.type.CommandClass;
@@ -75,22 +72,6 @@ private static void handleUnsupported(Command request, StreamObserver<Response>
         responseObserver.onError(Statuses.invalidArgumentWithCause(unsupported));
     }
 
-    @SuppressWarnings("RefusedBequest") // as we override default implementation with `unimplemented` status.
-    @Override
-    public void subscribe(Topic request, StreamObserver<Event> responseObserver) {
-        //TODO:2016-05-25:alexander.yevsyukov: Subscribe the client to the topic in the corresponding BoundedContext.
-        // This API is likely to change to support Firebase-like registration where listening is
-        // done by the client SDK implementation.
-    }
-
-    @SuppressWarnings("RefusedBequest") // as we override default implementation with `unimplemented` status.
-    @Override
-    public void unsubscribe(Topic request, StreamObserver<Response> responseObserver) {
-        //TODO:2016-05-25:alexander.yevsyukov: Unsubscribe the client from the topic in the corresponding BoundedContext.
-        responseObserver.onNext(Responses.ok());
-        responseObserver.onCompleted();
-    }
-
     public static class Builder {
 
         private final Set<BoundedContext> boundedContexts = Sets.newHashSet();

server/src/main/java/org/spine3/server/stand/Stand.java

@@ -258,7 +258,7 @@ public void execute(Query query, StreamObserver<QueryResponse> responseObserver)
 
     private ImmutableCollection<Any> internalExecute(Query query) {
 
-        final ImmutableSet.Builder<Any> resultBuilder = ImmutableSet.builder();
+        final ImmutableList.Builder<Any> resultBuilder = ImmutableList.builder();
 
         final Target target = query.getTarget();
 
@@ -280,7 +280,7 @@ private ImmutableCollection<Any> internalExecute(Query query) {
             feedStateRecordsToBuilder(resultBuilder, stateRecords);
         }
 
-        final ImmutableSet<Any> result = resultBuilder.build();
+        final ImmutableList<Any> result = resultBuilder.build();
 
         return result;
     }
@@ -293,7 +293,9 @@ private ImmutableCollection<EntityStorageRecord> fetchFromStandStorage(Target ta
 
         } else {
             final EntityFilters filters = target.getFilters();
-            if (filters != null && filters.getIdFilter() != null) {
+
+            // TODO[alex.tymchenko]: do we need to check for null at all? How about, say, Python gRPC client?
+            if (filters != null && filters.getIdFilter() != null && !filters.getIdFilter().getIdsList().isEmpty()) {
                 final EntityIdFilter idFilter = filters.getIdFilter();
                 final Collection<AggregateStateId> stateIds = Collections2.transform(idFilter.getIdsList(), aggregateStateIdTransformer(typeUrl));
 
@@ -355,15 +357,15 @@ private static ImmutableCollection<? extends Entity> fetchFromEntityRepository(T
         return result;
     }
 
-    private static void feedEntitiesToBuilder(ImmutableSet.Builder<Any> resultBuilder, ImmutableCollection<? extends Entity> all) {
+    private static void feedEntitiesToBuilder(ImmutableList.Builder<Any> resultBuilder, ImmutableCollection<? extends Entity> all) {
         for (Entity record : all) {
             final Message state = record.getState();
             final Any packedState = AnyPacker.pack(state);
             resultBuilder.add(packedState);
         }
     }
 
-    private static void feedStateRecordsToBuilder(ImmutableSet.Builder<Any> resultBuilder, ImmutableCollection<EntityStorageRecord> all) {
+    private static void feedStateRecordsToBuilder(ImmutableList.Builder<Any> resultBuilder, ImmutableCollection<EntityStorageRecord> all) {
         for (EntityStorageRecord record : all) {
             final Any state = record.getState();
             resultBuilder.add(state);

server/src/main/java/org/spine3/server/stand/StandFunnel.java

@@ -108,8 +108,6 @@ public Stand getStand() {
          *
          * <p> The value must not be null.
          *
-         * <p> If this method is not used, a default value will be used.
-         *
          * @param stand the instance of {@link Stand}.
          * @return {@code this} instance of {@code Builder}
          */
@@ -127,6 +125,8 @@ public Executor getExecutor() {
          *
          * <p>The value must not be {@code null}.
          *
+         * <p> If this method is not used, a default value will be used.
+         *
          * @param executor the instance of {@code Executor}.
          * @return {@code this} instance of {@code Builder}
          */