Merge pull request #10 from LukasPoque/add_s3i_event_system

Add S3I Event System

Author: LukasPoque

CHANGELOG.md

@@ -1,3 +1,13 @@
+## 0.5.0
+
+Add S3I Event System functionality:
+
+-  EventMessage, EventSubscriptionRequest, EventSubscriptionReply
+-  `POST` and `DELETE` methods for the S3I-Config-API
+-  Methods to create/remove normal and event broker endpoints (queues)
+-  Method to configure the `ActiveBrokerInterfaces` to be used as an event broker
+-  `EventSystemConnector` - a class to simplify the subscription to custom/named events
+
 ## 0.4.1
 
 Fix utf8 encoding problems.

README.md

@@ -47,6 +47,56 @@ Please see [pub.dev](https://pub.dev/packages/s3i_flutter/install) for instructi
 If you like this package, consider supporting it by giving a star on [GitHub](https://github.com/LukasPoque/s3i_flutter) and 
 a like on [pub.dev](https://pub.dev/packages/s3i_flutter) :heart:
 
+## Features
+
+The goal for `Version 1.0.0` is to cover the important endpoints of the main S³I components and provide useful data classes to wrap the JSON-data. For `Version 2.0.0` there should be more functionality to work with the data classes, especially with the (F)ML4.0 language.
+
+#### Roadmap to Version 1.0.0
+
+- Authentication
+  - [x] Authenticate an user via the S3I-OAuthProxy
+  - [x] Use an refresh/offline token for less user interaction during authentication
+  - [ ] Enable saving the refresh/offline token for authentication after a restart or offline
+  - [ ] Authenticate an user via default OpenId-Connect (with redirect url)
+
+- Directory
+  - [x] Provide basic PUT/GET/DELETE request methods
+  - [x] Request a single thing (with filter options)
+  - [x] Modify a single thing
+  - [x] Request a policy entry
+  - [x] Modify a policy entry
+  - [ ] Query the directory via thing search
+  - [ ] Find the owner of a thing
+  - [ ] Find all things that belongs to a specific person
+  - [ ] Create/delete a new thing in the S³I (adds a basic thing entry to the directory and creates a client in the identity provider)
+
+- Repository
+  - [ ] Provide basic PUT/GET/DELETE request methods
+  - [ ] Request a single thing (with filter options)
+  - [ ] Modify a single thing
+  - [ ] Request a policy entry
+  - [ ] Modify a policy entry
+  - [ ] Query the repository via thing search
+  - [ ] Create/delete a new thing entry in the repository
+  - [ ] Receive live updates from the cloud copy of a thing
+  - [ ] Send live updates to the cloud copy of a thing
+  
+- Messaging
+  - [x] Create/delete a new broker queue (bound to the direct exchange)
+  - [x] Receive/send messages using AMQP (not usable for web)
+  - [x] Receive/sent messages using the REST endpoint of the S3I-Broker-API
+  - [x] Work with UserMessages
+  - [x] Work with ServiceMessages
+  - [x] Work with GetValueMessages
+  - [ ] Work with SetValueMessages
+  - [ ] Work with DeleteAttributeMessages
+  - [ ] Work with CreateAttributeMessages
+  - [x] Work with messages from the EventSystem
+  - [x] Create/delete broker queues for the EventSystem
+  - [x] Simple to use wrapper for the EventSystem (as a subscriber)
+  - [ ] Helper functions for the EventSystem as publisher
+
+
 ## Usage
 
 For a basic example application see the [example](https://github.com/LukasPoque/s3i_flutter/tree/master/example).
@@ -175,7 +225,7 @@ brokerConnector.subscribeServiceReplyReceived((ServiceReply msg) {
 });
 ```
 
-If all callback you're interested in are registered it's time to start consuming on one (or multiple) queues on the S3I-Broker. 
+If all callbacks you're interested in are registered it's time to start consuming on one (or multiple) queues on the S3I-Broker. 
 For that simply call `startConsuming`. If you don't want any open connections left when your app closes, call `stopConsuming` with the same endpoint
 in your dispose method.
 ```dart
@@ -205,7 +255,7 @@ brokerConnector.sendMessage(requestMsg, <String>{'<SERVICE ENDPOINT>'});
 
 The package is divided in domain specific folders.
 
-TODO: ...
+The `S3ICore` uses this classes and provides methods to access the REST-APIs easier.
 
 ### Auth
 
@@ -231,8 +281,9 @@ At the moment, there are only active broker interfaces implemented:
 
 Currently the following message types are supported:
 - `UserMessage`: used for communication between two real users.
-- `ServiceMessage`: used to invoke service functions or receive service answers from S3I-Services.
-- `GetValueMessage`: used to get a specific value from an other thing.
+- `ServiceMessages`: used to invoke service functions or receive service answers from S3I-Services.
+- `GetValueMessages`: used to get a specific value from an other thing.
+- `EventSystemMessages`: used to receive/subscribe to events via the [S3I-Event-System](https://github.com/LukasPoque/s3i_flutter/issues/9#issuecomment-925665563).
 
 ### Directory
 

example/pubspec.lock

@@ -7,7 +7,7 @@ packages:
       name: async
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "2.6.1"
+    version: "2.8.1"
   boolean_selector:
     dependency: transitive
     description:
@@ -28,7 +28,7 @@ packages:
       name: charcode
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "1.2.0"
+    version: "1.3.1"
   clock:
     dependency: transitive
     description:
@@ -120,7 +120,7 @@ packages:
       name: meta
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "1.3.0"
+    version: "1.7.0"
   path:
     dependency: transitive
     description:
@@ -148,7 +148,7 @@ packages:
       path: ".."
       relative: true
     source: path
-    version: "0.4.0"
+    version: "0.5.0"
   sky_engine:
     dependency: transitive
     description: flutter
@@ -195,7 +195,7 @@ packages:
       name: test_api
       url: "https://pub.dartlang.org"
     source: hosted
-    version: "0.3.0"
+    version: "0.4.2"
   typed_data:
     dependency: transitive
     description:

lib/s3i_flutter.dart

@@ -7,12 +7,12 @@ export 'src/auth/oauth_proxy_flow.dart';
 export 'src/auth/tokens.dart';
 
 //broker
-export 'src/broker/attribute_value_messages.dart';
+export 'src/broker/broker_amqp_connector.dart'
+    if (dart.library.js) 'src/broker/broker_rest_connector.dart';
 export 'src/broker/broker_interfaces.dart';
-export 'src/broker/broker_rest_connector.dart'
-    if (dart.library.io) 'src/broker/broker_amqp_connector.dart';
-export 'src/broker/service_messages.dart';
-export 'src/broker/user_message.dart';
+export 'src/broker/messages/attribute_value_messages.dart';
+export 'src/broker/messages/service_messages.dart';
+export 'src/broker/messages/user_message.dart';
 
 //directory
 export 'src/directory/dir_object.dart';

lib/src/broker/broker_amqp_connector.dart

@@ -4,7 +4,7 @@ import 'package:dart_amqp/dart_amqp.dart';
 import 'package:s3i_flutter/src/auth/authentication_manager.dart';
 import 'package:s3i_flutter/src/auth/tokens.dart';
 import 'package:s3i_flutter/src/broker/broker_interfaces.dart';
-import 'package:s3i_flutter/src/broker/message.dart';
+import 'package:s3i_flutter/src/broker/messages/message.dart';
 import 'package:s3i_flutter/src/exceptions/network_authentication_exception.dart';
 import 'package:s3i_flutter/src/exceptions/s3i_exception.dart';
 
@@ -15,6 +15,16 @@ import 'package:s3i_flutter/src/exceptions/s3i_exception.dart';
 /// parameters).
 ///
 /// This is needed to use the same interface on web and other platforms.
+///
+/// The [args] have defaults for the normal message communication in the S3I:
+/// - brokerHost = 'rabbitmq.s3i.vswf.dev',
+/// - port = 5672,
+/// - virtualHost = 's3i',
+/// - maxConnectionAttempts = 3,
+/// - reconnectWaitTime = const Duration(milliseconds: 1500),
+/// - prefetchCount = -1,
+/// - exchangeName = 'demo.direct',
+/// - exchangeType = ExchangeType.DIRECT
 ActiveBrokerInterface getActiveBrokerDefaultConnector(
     AuthenticationManager authManager,
     {Map<String, dynamic> args = const <String, dynamic>{}}) {
@@ -25,14 +35,33 @@ ActiveBrokerInterface getActiveBrokerDefaultConnector(
   final int maxConnectionAttempts = args['maxConnectionAttempts'] as int? ?? 3;
   final Duration reconnectWaitTime = args['reconnectWaitTime'] as Duration? ??
       const Duration(milliseconds: 1500);
+  final int prefetchCount = args['prefetchCount'] as int? ?? -1;
   final String exchangeName = args['exchangeName'] as String? ?? 'demo.direct';
+  final ExchangeType exchangeType =
+      args['exchangeType'] as ExchangeType? ?? ExchangeType.DIRECT;
+
   return BrokerAmqpConnector(authManager,
       brokerHost: brokerHost,
       port: port,
       virtualHost: virtualHost,
       maxConnectionAttempts: maxConnectionAttempts,
       reconnectWaitTime: reconnectWaitTime,
-      exchangeName: exchangeName);
+      prefetchCount: prefetchCount,
+      exchangeName: exchangeName,
+      exchangeType: exchangeType);
+}
+
+/// Creates a new [BrokerAmqpConnector] configured to be used for the
+/// Event System.
+///
+/// See [getActiveBrokerDefaultConnector] for more information.
+ActiveBrokerInterface getActiveBrokerEventConnector(
+    AuthenticationManager authManager,
+    {Map<String, dynamic> args = const <String, dynamic>{}}) {
+  if (args.containsKey('exchangeName')) args['exchangeName'] = 'eventExchange';
+  if (args.containsKey('exchangeType'))
+    args['exchangeType'] = ExchangeType.TOPIC;
+  return getActiveBrokerDefaultConnector(authManager, args: args);
 }
 
 /// This [ActiveBrokerInterface] implementation uses the native messaging
@@ -45,16 +74,15 @@ ActiveBrokerInterface getActiveBrokerDefaultConnector(
 class BrokerAmqpConnector extends ActiveBrokerInterface {
   /// Creates a [BrokerAmqpConnector] which uses the [authManager] to receive
   /// tokens.
-  ///
-  /// Unless you know what you are doing, don't change the default values
-  /// of the named parameters.
   BrokerAmqpConnector(AuthenticationManager authManager,
-      {this.brokerHost = 'rabbitmq.s3i.vswf.dev',
-      this.port = 5672,
-      this.virtualHost = 's3i',
-      this.maxConnectionAttempts = 3,
-      this.reconnectWaitTime = const Duration(milliseconds: 1500),
-      this.exchangeName = 'demo.direct'})
+      {required this.brokerHost,
+      required this.port,
+      required this.virtualHost,
+      required this.maxConnectionAttempts,
+      required this.reconnectWaitTime,
+      required this.prefetchCount,
+      required this.exchangeName,
+      required this.exchangeType})
       : super(authManager);
 
   /// The host to connect to.
@@ -72,9 +100,19 @@ class BrokerAmqpConnector extends ActiveBrokerInterface {
   /// The time between each reconnect attempt.
   final Duration reconnectWaitTime;
 
+  /// The number of messages loaded to the client without ACK.
+  ///
+  /// If this number is < 0 no prefetch count is set => infinite prefetch.
+  final int prefetchCount;
+
   /// The name of the amqp broker exchange.
   final String exchangeName;
 
+  /// The type of the exchange, see
+  /// [here](https://www.rabbitmq.com/tutorials/amqp-concepts.html#exchanges)
+  /// for more information.
+  final ExchangeType exchangeType;
+
   /// Stores all endpoints and their consumer queues.
   final Map<String, Consumer> _endpointConsumer = <String, Consumer>{};
 
@@ -215,8 +253,11 @@ class BrokerAmqpConnector extends ActiveBrokerInterface {
     if (_amqpClient == null) throw S3IException('amqp client is null');
     try {
       _channel = await _amqpClient!.channel();
-      _exchange = await _channel!
-          .exchange(exchangeName, ExchangeType.DIRECT, passive: true);
+      if (prefetchCount >= 0) {
+        _channel = await _channel!.qos(null, prefetchCount);
+      }
+      _exchange =
+          await _channel!.exchange(exchangeName, exchangeType, passive: true);
     } on Exception catch (e) {
       throw S3IException('amqp package error: $e');
     }