Merge pull request #35 from SpineEventEngine/async-subscriptions

Async subscriptions

Author: dmdashenkov

client/lib/client.dart

@@ -250,7 +250,7 @@ class Client {
     void cancelAllSubscriptions() {
         for (Subscription subscription in _activeSubscriptions) {
             subscription.unsubscribe();
-            subscription.subscription.then(_cancel);
+            _cancel(subscription.subscription);
         }
         _activeSubscriptions.clear();
     }
@@ -266,19 +266,19 @@ class Client {
         });
     }
 
-    EventSubscription<E>
+    Future<EventSubscription<E>>
     _subscribeToEvents<E extends GeneratedMessage>(pbSubscription.Topic topic) {
         return _subscribe(topic, (s, d) => EventSubscription.of(s, d));
     }
 
-    StateSubscription<S>
+    Future<StateSubscription<S>>
     _subscribeToStateUpdates<S extends GeneratedMessage>(pbSubscription.Topic topic,
                                                          BuilderInfo builderInfo) {
         return _subscribe(topic, (s, d) => StateSubscription.of(s, builderInfo, d));
     }
 
-    S _subscribe<S extends Subscription>(pbSubscription.Topic topic,
-                                         _CreateSubscription<S> newSubscription) {
+    Future<S> _subscribe<S extends Subscription>(pbSubscription.Topic topic,
+                                                 _CreateSubscription<S> newSubscription) {
         if (_firebase == null) {
             throw StateError('Cannot create a subscription. No Firebase client is provided.');
         }
@@ -287,10 +287,11 @@ class Client {
         if (builder == null) {
             throw ArgumentError.value(topic, 'topic', 'Target type `$targetTypeUrl` is unknown.');
         }
-        var fbSubscription = _httpClient
+        var subscription = _httpClient
             .postMessage(_endpoints.subscription.create, topic)
-            .then(_parseFirebaseSubscription);
-        return newSubscription(fbSubscription, _firebase!);
+            .then(_parseFirebaseSubscription)
+            .then((value) => newSubscription(value, _firebase!));
+        return subscription;
     }
 
     FirebaseSubscription _parseFirebaseSubscription(http.Response response) {
@@ -311,10 +312,10 @@ class Client {
     void _refreshSubscription(Subscription subscription) {
         var subscriptionMessage = subscription.subscription;
         if (subscription.closed) {
-            subscriptionMessage.then(_cancel);
+            _cancel(subscriptionMessage);
             _activeSubscriptions.remove(subscription);
         } else {
-            subscriptionMessage.then(_keepUp);
+            _keepUp(subscriptionMessage);
         }
     }
 
@@ -330,69 +331,111 @@ class Client {
 /// A function which accepts a future of `FirebaseSubscription` and a firebase client and creates
 /// an instance of [Subscription].
 typedef _CreateSubscription<S extends Subscription> =
-    S Function(Future<FirebaseSubscription>, FirebaseClient);
+    S Function(FirebaseSubscription, FirebaseClient);
+
+/// A simple or a composite field filter.
+///
+abstract class FilterOrComposite {
+
+    /// Obtains the Protobuf `CompositeFilter` representing this filter.
+    CompositeFilter _toProto();
+}
+
+/// A composite field filter.
+///
+class Composite implements FilterOrComposite {
+
+    final CompositeFilter filter;
+
+    Composite._(this.filter);
+
+    @override
+    CompositeFilter _toProto() {
+        return filter;
+    }
+}
+
+/// A simple field filter.
+///
+/// A single simple filter is represented as a `CompositeFilter` by wrapping it into the composite
+/// filter with the `AND` operator.
+///
+class SimpleFilter implements FilterOrComposite {
+
+    final Filter filter;
+
+    SimpleFilter._(this.filter);
+
+    @override
+    CompositeFilter _toProto() {
+        return CompositeFilter()
+            ..filter.add(filter)
+            ..operator = CompositeFilter_CompositeOperator.ALL
+            ..freeze();
+    }
+}
 
 /// Creates a composite filter which groups one or more field filters with the `ALL` operator.
 ///
 /// All the field filters should pass in order for the composite filter to pass.
 ///
-CompositeFilter all(Iterable<Filter> filters) {
+Composite all(Iterable<SimpleFilter> filters) {
     ArgumentError.checkNotNull(filters);
-    return CompositeFilter()
+    return Composite._(CompositeFilter()
         ..operator = CompositeFilter_CompositeOperator.ALL
-        ..filter.addAll(filters)
-        ..freeze();
+        ..filter.addAll(filters.map((f) => f.filter))
+        ..freeze());
 }
 
 /// Creates a composite filter which groups one or more field filters with the `EITHER` operator.
 ///
 /// At least one field filter should pass in order for the composite filter to pass.
 ///
-CompositeFilter either(Iterable<Filter> filters) {
+Composite either(Iterable<SimpleFilter> filters) {
     ArgumentError.checkNotNull(filters);
-    return CompositeFilter()
+    return Composite._(CompositeFilter()
         ..operator = CompositeFilter_CompositeOperator.EITHER
-        ..filter.addAll(filters)
-        ..freeze();
+        ..filter.addAll(filters.map((f) => f.filter))
+        ..freeze());
 }
 
 /// Creates a field filter with the `=` operator.
-Filter eq(String fieldPath, Object value) =>
+SimpleFilter eq(String fieldPath, Object value) =>
     _filter(fieldPath, Filter_Operator.EQUAL, value);
 
 /// Creates a field filter with the `<=` operator.
-Filter le(String fieldPath, Object value) =>
+SimpleFilter le(String fieldPath, Object value) =>
     _filter(fieldPath, Filter_Operator.LESS_OR_EQUAL, value);
 
 /// Creates a field filter with the `>=` operator.
-Filter ge(String fieldPath, Object value) =>
+SimpleFilter ge(String fieldPath, Object value) =>
     _filter(fieldPath, Filter_Operator.GREATER_OR_EQUAL, value);
 
 /// Creates a field filter with the `<` operator.
-Filter lt(String fieldPath, Object value) =>
+SimpleFilter lt(String fieldPath, Object value) =>
     _filter(fieldPath, Filter_Operator.LESS_THAN, value);
 
 /// Creates a field filter with the `>` operator.
-Filter gt(String fieldPath, Object value) =>
+SimpleFilter gt(String fieldPath, Object value) =>
     _filter(fieldPath, Filter_Operator.GREATER_THAN, value);
 
-Filter _filter(String fieldPath, Filter_Operator operator, Object value) {
+SimpleFilter _filter(String fieldPath, Filter_Operator operator, Object value) {
     ArgumentError.checkNotNull(fieldPath);
     ArgumentError.checkNotNull(value);
     var pathElements = fieldPath.split('.');
-    return Filter()
+    return SimpleFilter._(Filter()
         ..fieldPath = (FieldPath()..fieldName.addAll(pathElements))
         ..operator = operator
         ..value = packObject(value)
-        ..freeze();
+        ..freeze());
 }
 
 /// A request to the server to post a command.
 class CommandRequest<M extends GeneratedMessage> {
 
     final Client _client;
     final Command _command;
-    final List<EventSubscription> _subscriptions = [];
+    final List<Future<EventSubscription>> _futureSubscriptions = [];
 
     CommandRequest._(this._client, M command) :
             _command = _client._requests.command().create(command);
@@ -402,11 +445,15 @@ class CommandRequest<M extends GeneratedMessage> {
     /// Events down the line, i.e. events produced as the result of other messages which where
     /// produced as the result of this command, do not match this subscription.
     ///
-    EventSubscription<E> observeEvents<E extends GeneratedMessage>() {
+    /// When the resulting future completes, the subscription is guaranteed to be created.
+    /// Also, when the future created in `post(..)` completes, the all subscriptions created within
+    /// the same `CommandRequest` are guaranteed to have completed.
+    ///
+    Future<EventSubscription<E>> observeEvents<E extends GeneratedMessage>() {
         var subscription = _client.subscribeToEvents<E>()
-            .where(all([eq('context.past_message', _commandAsOrigin())]))
+            .where(eq('context.past_message', _commandAsOrigin()))
             .post();
-        _subscriptions.add(subscription);
+        _futureSubscriptions.add(subscription);
         return subscription;
     }
 
@@ -424,19 +471,22 @@ class CommandRequest<M extends GeneratedMessage> {
     /// Fails if there are no event subscriptions to monitor the command execution. If this is
     /// the desired behaviour, use `CommandRequest.postAndForget(..)`.
     ///
+    /// When the command is sent, the event subscriptions created within this request
+    /// are guaranteed to be active.
+    ///
     /// Returns a future which completes when the request is sent. If there was a network problem,
     /// the future, completes with an error.
     ///
     /// If the server rejects the command with an error and the [onError] callback is set,
     /// the callback will be triggered with the error. Otherwise, the error is silently ignored.
     ///
     Future<void> post({CommandErrorCallback? onError}) {
-        if (_subscriptions.isEmpty) {
+        if (_futureSubscriptions.isEmpty) {
             throw StateError('Use `observeEvents(..)` or `observeEventsWithContexts(..)` to observe'
                 ' command results or call `postAndForget()` instead of `post()` if you observe'
                 ' command results elsewhere.');
         }
-        return Future.wait(_subscriptions.map((s) => s.subscription))
+        return Future.wait(_futureSubscriptions)
                      .then((_) => _client._postCommand(_command, onError));
     }
 
@@ -452,7 +502,7 @@ class CommandRequest<M extends GeneratedMessage> {
     /// the callback will be triggered with the error. Otherwise, the error is silently ignored.
     ///
     Future<void> postAndForget({CommandErrorCallback? onError}) {
-        if (_subscriptions.isNotEmpty) {
+        if (_futureSubscriptions.isNotEmpty) {
             throw StateError('Use `post()` to add event subscriptions.');
         }
         return _client._postCommand(_command, onError);
@@ -502,9 +552,9 @@ class QueryRequest<M extends GeneratedMessage> {
     /// an entity state should pass all of the composite filters to be included in the query
     /// results.
     ///
-    QueryRequest<M> where(CompositeFilter filter) {
+    QueryRequest<M> where(FilterOrComposite filter) {
         ArgumentError.checkNotNull(filter, 'filter');
-        _filters.add(filter);
+        _filters.add(filter._toProto());
         return this;
     }
 
@@ -585,9 +635,9 @@ class StateSubscriptionRequest<M extends GeneratedMessage> {
     /// If called multiple times, the composite filters are composed with the `ALL` operator, i.e.
     /// an entity state should pass all of the composite filters to match the subscription.
     ///
-    StateSubscriptionRequest<M> where(CompositeFilter filter) {
+    StateSubscriptionRequest<M> where(FilterOrComposite filter) {
         ArgumentError.checkNotNull(filter, 'filter');
-        _filters.add(filter);
+        _filters.add(filter._toProto());
         return this;
     }
 
@@ -605,7 +655,10 @@ class StateSubscriptionRequest<M extends GeneratedMessage> {
 
     /// Asynchronously sends this request to the server.
     ///
-    StateSubscription<M> post() {
+    /// The subscription is guaranteed to have been created on server when the resulting future
+    /// completes.
+    ///
+    Future<StateSubscription<M>> post() {
         var topic = _client._requests.topic().withFilters(_type, ids: _ids, filters: _filters);
         var builderInfo = theKnownTypes.findBuilderInfo(theKnownTypes.typeUrlFrom(_type))!;
         return _client._subscribeToStateUpdates(topic, builderInfo);
@@ -629,15 +682,18 @@ class EventSubscriptionRequest<M extends GeneratedMessage> {
     /// If called multiple times, the composite filters are composed with the `ALL` operator, i.e.
     /// an event should pass all of the composite filters to match the subscription.
     ///
-    EventSubscriptionRequest<M> where(CompositeFilter filter) {
+    EventSubscriptionRequest<M> where(FilterOrComposite filter) {
         ArgumentError.checkNotNull(filter, 'filter');
-        _filers.add(filter);
+        _filers.add(filter._toProto());
         return this;
     }
 
     /// Asynchronously sends this request to the server.
     ///
-    EventSubscription<M> post() {
+    /// The subscription is guaranteed to have been created on server when the resulting future
+    /// completes.
+    ///
+    Future<EventSubscription<M>> post() {
         var topic = _client._requests.topic().withFilters(_type, filters: _filers);
         return _client._subscribeToEvents(topic);
     }

client/lib/subscription.dart

@@ -41,20 +41,15 @@ class Subscription<T extends GeneratedMessage> {
     ///
     /// Completes when the subscription is created.
     ///
-    final Future<pb.Subscription> subscription;
+    final pb.Subscription subscription;
 
     final Stream<T> _itemAdded;
 
     bool _closed;
 
     Subscription._(this.subscription, Stream<T> itemAdded)
         : _itemAdded = _checkBroadcast(itemAdded),
-          _closed = false {
-        subscription.catchError((e) {
-            unsubscribe();
-            return pb.Subscription.getDefault();
-        });
-    }
+          _closed = false;
 
     /// Shows if this subscription is still active or already closed.
     bool get closed => _closed;
@@ -86,7 +81,7 @@ class StateSubscription<T extends GeneratedMessage> extends Subscription<T> {
 
     bool _closed;
 
-    StateSubscription._(Future<pb.Subscription> subscription,
+    StateSubscription._(pb.Subscription subscription,
                         Stream<T> itemAdded,
                         Stream<T> itemChanged,
                         Stream<T> itemRemoved):
@@ -96,19 +91,18 @@ class StateSubscription<T extends GeneratedMessage> extends Subscription<T> {
             super._(subscription, itemAdded);
 
     /// Creates a new instance which broadcasts updates from under the given Firebase node.
-    factory StateSubscription.of(Future<FirebaseSubscription> firebaseSubscription,
+    factory StateSubscription.of(FirebaseSubscription firebaseSubscription,
                                  BuilderInfo builderInfoForType,
                                  FirebaseClient database) {
-        var subscription = firebaseSubscription.then((value) => value.subscription);
-        var nodePath = _nodePath(firebaseSubscription);
-        var itemAdded = _nodePathStream(nodePath)
-            .asyncExpand((element) => database.childAdded(element))
+        var subscription = firebaseSubscription.subscription;
+        var nodePath = firebaseSubscription.nodePath.value;
+        var itemAdded = database
+            .childAdded(nodePath)
             .map((json) => parseIntoNewInstance<T>(builderInfoForType, json));
-        var itemChanged = _nodePathStream(nodePath)
-            .asyncExpand((element) => database.childChanged(element))
+        var itemChanged = database
+            .childChanged(nodePath)
             .map((json) => parseIntoNewInstance<T>(builderInfoForType, json));
-        var itemRemoved = _nodePathStream(nodePath)
-            .asyncExpand((element) => database.childRemoved(element))
+        var itemRemoved = database.childRemoved(nodePath)
             .map((json) => parseIntoNewInstance<T>(builderInfoForType, json));
         return StateSubscription._(subscription, itemAdded, itemChanged, itemRemoved);
     }
@@ -127,14 +121,14 @@ class EventSubscription<T extends GeneratedMessage> extends Subscription<Event>
 
     static final BuilderInfo _eventBuilderInfo = Event.getDefault().info_;
 
-    EventSubscription._(Future<pb.Subscription> subscription, Stream<Event> itemAdded) :
+    EventSubscription._(pb.Subscription subscription, Stream<Event> itemAdded) :
             super._(subscription, itemAdded);
 
-    factory EventSubscription.of(Future<FirebaseSubscription> firebaseSubscription,
+    factory EventSubscription.of(FirebaseSubscription firebaseSubscription,
                                  FirebaseClient database) {
-        var subscription = firebaseSubscription.then((value) => value.subscription);
-        var itemAdded = _nodePathStream(_nodePath(firebaseSubscription))
-            .asyncExpand((path) => database.childAdded(path))
+        var subscription = firebaseSubscription.subscription;
+        var nodePath = firebaseSubscription.nodePath.value;
+        var itemAdded = database.childAdded(nodePath)
             .map((json) => parseIntoNewInstance<Event>(_eventBuilderInfo, json));
         return EventSubscription._(subscription, itemAdded);
     }
@@ -147,13 +141,6 @@ class EventSubscription<T extends GeneratedMessage> extends Subscription<Event>
         .map((event) => unpack(event.message) as T);
 }
 
-Future<String> _nodePath(Future<FirebaseSubscription> subscription) =>
-    subscription
-        .then((value) => value.nodePath.value);
-
-Stream<String> _nodePathStream(Future<String> futureValue) =>
-    futureValue.asStream().asBroadcastStream();
-
 Stream<T> _checkBroadcast<T>(Stream<T> stream) {
     if (!stream.isBroadcast) {
         throw ArgumentError(