Update

PlugFox · Jul 23, 2023 · f38b98f · f38b98f
1 parent 3a74832
commit f38b98f
Show file tree

Hide file tree

Showing 11 changed files with 499 additions and 44 deletions.
diff --git a/.vscode/tasks.json b/.vscode/tasks.json
@@ -14,21 +14,60 @@
     {
       "label": "Start Centrifugo Server",
       "type": "shell",
-      "command": "docker",
-      "args": [
-        "run",
-        "-d",
-        "--rm",
-        "--ulimit=nofile=65536:65536",
-        "-p=8000:8000",
-        "--name=centrifugo",
-        "centrifugo/centrifugo:latest",
-        "centrifugo",
-        "--client_insecure",
-        "--admin",
-        "--admin_insecure",
-        "--log_level=debug"
-      ],
+      "windows": {
+        "command": "docker",
+        "args": [
+          "run",
+          "-d",
+          "--rm",
+          "--ulimit=nofile=65536:65536",
+          "-p=8000:8000/tcp",
+          "--volume=${PWD}/config.json:/centrifugo/config.json:ro",
+          "--name=centrifugo",
+          "centrifugo/centrifugo:latest",
+          "centrifugo",
+          //"--client_insecure",
+          "--admin",
+          "--admin_insecure",
+          "--log_level=debug"
+        ]
+      },
+      "linux": {
+        "command": "docker",
+        "args": [
+          "run",
+          "-d",
+          "--rm",
+          "--ulimit=nofile=65536:65536",
+          "-p=8000:8000/tcp",
+          "--volume=${PWD}/config.json:/centrifugo/config.json:ro",
+          "--name=centrifugo",
+          "centrifugo/centrifugo:latest",
+          "centrifugo",
+          //"--client_insecure",
+          "--admin",
+          "--admin_insecure",
+          "--log_level=debug"
+        ]
+      },
+      "osx": {
+        "command": "docker",
+        "args": [
+          "run",
+          "-d",
+          "--rm",
+          "--ulimit=nofile=65536:65536",
+          "-p=8000:8000/tcp",
+          "--volume=${PWD}/config.json:/centrifugo/config.json:ro",
+          "--name=centrifugo",
+          "centrifugo/centrifugo:latest",
+          "centrifugo",
+          "--client_insecure",
+          "--admin",
+          "--admin_insecure",
+          "--log_level=debug"
+        ]
+      },
       "group": {
         "kind": "none",
         "isDefault": true
@@ -53,6 +92,32 @@
         "reveal": "always",
         "panel": "dedicated"
       }
+    },
+    {
+      "label": "Generate new user token",
+      "type": "shell",
+      "command": "docker",
+      "args": [
+        "run",
+        "-it",
+        "--rm",
+        "--volume=${PWD}/config.json:/centrifugo/config.json:ro",
+        "--name=centrifugo-cli",
+        "centrifugo/centrifugo:latest",
+        "centrifugo",
+        "gentoken",
+        "--user=dart",
+        "--ttl=604800000" // 604800
+      ],
+      "group": {
+        "kind": "none",
+        "isDefault": true
+      },
+      "problemMatcher": [],
+      "presentation": {
+        "reveal": "always",
+        "panel": "dedicated"
+      }
     }
   ]
 }
diff --git a/examples/console/README.md b/examples/console/README.md
@@ -89,15 +89,15 @@ Bash:
 ```bash
 docker run -it --rm --volume ${PWD}/config.json:/centrifugo/config.json:ro \
     --name centrifugo-cli centrifugo/centrifugo:latest \
-    centrifugo gensubtoken --user dart --channel chat:index
+    centrifugo gentoken --user dart
 ```
 
 PowerShell:
 
 ```powershell
 docker run -it --rm --volume ${PWD}/config.json:/centrifugo/config.json:ro `
     --name centrifugo-cli centrifugo/centrifugo:latest `
-    centrifugo gensubtoken --user dart --channel chat:index
+    centrifugo gentoken --user dart
 ```
 
 ### Run Centrifugo

diff --git a/examples/console/bin/main.dart b/examples/console/bin/main.dart
@@ -6,22 +6,44 @@ import 'dart:io' as io show exit, Platform;
 import 'package:args/args.dart' show ArgParser;
 import 'package:centrifuge_dart/centrifuge.dart';
 
+const url = 'ws://localhost:8000/connection/websocket?format=protobuf';
+
 void main([List<String>? args]) {
   final options = _extractOptions(args ?? const <String>[]);
   runZonedGuarded<void>(
     () async {
+      // Create centrifuge client.
       final client = Centrifuge(
         CentrifugeConfig(
           client: (
             name: 'Centrifuge Console Example',
             version: '0.0.1',
           ),
+          getToken: () =>
+              'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiJkYXJ0IiwiZXhwIj'
+              'oyMjk0OTE1MTMyLCJpYXQiOjE2OTAxMTUxMzJ9.hIGDXKn-eMdsdj57wn6-4y5p'
+              'k0tZcKoJCu0qxuuWSoQ',
         ),
       );
-      await client
-          .connect('ws://localhost:8000/connection/websocket?format=protobuf');
 
-      await Future<void>.delayed(const Duration(seconds: 3));
+      // Connect to centrifuge server using provided URL.
+      await client.connect(url);
+
+      // Output current client state.
+      print('Current state after connect: ${client.state}');
+
+      // State changes.
+      // Or you can observe specific state changes.
+      // e.g. `client.states.connected`
+      client.states.listen((state) => print('State changed to: $state'));
+
+      // Handle all centrifuge errors.
+      client.errors.listen(
+        (error) => print(
+          'Exception: ${error.exception}, '
+          'Stack trace: ${error.stackTrace}',
+        ),
+      );
 
       // TODO(plugfox): Read from stdin and send to channel.
 

diff --git a/lib/centrifuge.dart b/lib/centrifuge.dart
@@ -1,7 +1,8 @@
 library centrifuge;
 
-export 'src/client/centrifuge.dart' show Centrifuge;
-export 'src/model/config.dart';
-export 'src/model/exception.dart';
-export 'src/model/jwt.dart';
-export 'src/model/state.dart';
+export 'package:centrifuge_dart/src/client/centrifuge.dart' show Centrifuge;
+export 'package:centrifuge_dart/src/model/config.dart';
+export 'package:centrifuge_dart/src/model/exception.dart';
+export 'package:centrifuge_dart/src/model/jwt.dart';
+export 'package:centrifuge_dart/src/model/state.dart';
+export 'package:centrifuge_dart/src/model/states_stream.dart';
diff --git a/lib/src/client/centrifuge.dart b/lib/src/client/centrifuge.dart
@@ -4,6 +4,7 @@ import 'package:centrifuge_dart/src/client/centrifuge_interface.dart';
 import 'package:centrifuge_dart/src/model/config.dart';
 import 'package:centrifuge_dart/src/model/exception.dart';
 import 'package:centrifuge_dart/src/model/state.dart';
+import 'package:centrifuge_dart/src/model/states_stream.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:centrifuge_dart/src/transport/ws_protobuf_transport.dart';
 import 'package:centrifuge_dart/src/util/logger.dart' as logger;
@@ -12,7 +13,8 @@ import 'package:meta/meta.dart';
 /// {@template centrifuge}
 /// Centrifuge client.
 /// {@endtemplate}
-final class Centrifuge extends CentrifugeBase with CentrifugeConnectionMixin {
+final class Centrifuge extends CentrifugeBase
+    with CentrifugeErrorsMixin, CentrifugeConnectionMixin {
   /// {@macro centrifuge}
   Centrifuge([CentrifugeConfig? config])
       : super(config ?? CentrifugeConfig.defaultConfig());
@@ -45,7 +47,8 @@ abstract base class CentrifugeBase implements ICentrifuge {
   CentrifugeState get state => _transport.state;
 
   @override
-  Stream<CentrifugeState> get states => _transport.states;
+  late final CentrifugeStatesStream states =
+      CentrifugeStatesStream(_transport.states);
 
   /// Centrifuge config.
   /// {@nodoc}
@@ -67,19 +70,44 @@ abstract base class CentrifugeBase implements ICentrifuge {
 /// Mixin responsible for connection.
 /// {@nodoc}
 @internal
-base mixin CentrifugeConnectionMixin on CentrifugeBase {
+base mixin CentrifugeErrorsMixin on CentrifugeBase {
+  @protected
+  @nonVirtual
+  void _emitError(CentrifugeException exception, StackTrace stackTrace) =>
+      _errorsController.add((exception: exception, stackTrace: stackTrace));
+
+  late final StreamController<
+          ({CentrifugeException exception, StackTrace stackTrace})>
+      _errorsController = StreamController<
+          ({CentrifugeException exception, StackTrace stackTrace})>.broadcast();
+
+  @override
+  late final Stream<({CentrifugeException exception, StackTrace stackTrace})>
+      errors = _errorsController.stream;
+
+  @override
+  Future<void> close() async {
+    await super.close();
+    _errorsController.close().ignore();
+  }
+}
+
+/// Mixin responsible for connection.
+/// {@nodoc}
+@internal
+base mixin CentrifugeConnectionMixin on CentrifugeBase, CentrifugeErrorsMixin {
   @override
   Future<void> connect(String url) async {
     logger.fine('Interactively connecting to $url');
     try {
       await _transport.connect(url);
-    } on CentrifugeException {
+    } on CentrifugeException catch (error, stackTrace) {
+      _emitError(error, stackTrace);
       rethrow;
     } on Object catch (error, stackTrace) {
-      Error.throwWithStackTrace(
-        CentrifugeConnectionException(error),
-        stackTrace,
-      );
+      final centrifugeException = CentrifugeConnectionException(error);
+      _emitError(centrifugeException, stackTrace);
+      Error.throwWithStackTrace(centrifugeException, stackTrace);
     }
   }
 
@@ -88,13 +116,13 @@ base mixin CentrifugeConnectionMixin on CentrifugeBase {
     logger.fine('Interactively disconnecting');
     try {
       await _transport.disconnect(0, 'Disconnect called');
-    } on CentrifugeException {
+    } on CentrifugeException catch (error, stackTrace) {
+      _emitError(error, stackTrace);
       rethrow;
     } on Object catch (error, stackTrace) {
-      Error.throwWithStackTrace(
-        CentrifugeDisconnectionException(error),
-        stackTrace,
-      );
+      final centrifugeException = CentrifugeConnectionException(error);
+      _emitError(centrifugeException, stackTrace);
+      Error.throwWithStackTrace(centrifugeException, stackTrace);
     }
   }
 

diff --git a/lib/src/client/centrifuge_interface.dart b/lib/src/client/centrifuge_interface.dart
@@ -1,12 +1,16 @@
-import 'package:centrifuge_dart/src/model/state.dart';
+import 'package:centrifuge_dart/centrifuge.dart';
 
 /// Centrifuge client interface.
 abstract interface class ICentrifuge {
   /// State of client.
   CentrifugeState get state;
 
   /// Stream of client states.
-  abstract final Stream<CentrifugeState> states;
+  abstract final CentrifugeStatesStream states;
+
+  /// Stream of errors.
+  abstract final Stream<
+      ({CentrifugeException exception, StackTrace stackTrace})> errors;
 
   /* abstract final Stream<Object> publications; */
 

diff --git a/lib/src/model/state.dart b/lib/src/model/state.dart
@@ -1,3 +1,5 @@
+import 'dart:convert';
+
 import 'package:meta/meta.dart';
 
 /// {@template state}
@@ -29,6 +31,7 @@ import 'package:meta/meta.dart';
 /// Also, this can happen due to server advice from a server,
 /// or due to a terminal problem that happened on the client-side.
 /// {@endtemplate}
+@immutable
 sealed class CentrifugeState extends _$CentrifugeStateBase {
   /// {@macro state}
   const CentrifugeState(super.timestamp);
@@ -59,6 +62,7 @@ sealed class CentrifugeState extends _$CentrifugeStateBase {
     bool? sendPong,
     String? session,
     String? node,
+    List<int>? data,
   }) = CentrifugeState$Connected;
 
   /// Permanently closed
@@ -111,6 +115,10 @@ sealed class CentrifugeState extends _$CentrifugeStateBase {
             },
             session: json['session']?.toString(),
             node: json['node']?.toString(),
+            data: switch (json['data']) {
+              String data when data.isNotEmpty => base64Decode(data),
+              _ => null,
+            },
           ),
         ('closed', int timestamp, _) => CentrifugeState.closed(
             timestamp: DateTime.fromMicrosecondsSinceEpoch(timestamp),
@@ -261,6 +269,7 @@ final class CentrifugeState$Connected extends CentrifugeState
     this.sendPong,
     this.session,
     this.node,
+    this.data,
   }) : super(timestamp ?? DateTime.now());
 
   @override
@@ -296,6 +305,9 @@ final class CentrifugeState$Connected extends CentrifugeState
   /// Server node ID.
   final String? node;
 
+  /// Additional data returned from server on connect.
+  final List<int>? data;
+
   @override
   bool get isDisconnected => false;
 
@@ -328,6 +340,7 @@ final class CentrifugeState$Connected extends CentrifugeState
         if (sendPong != null) 'sendPong': sendPong,
         if (session != null) 'session': session,
         if (node != null) 'node': node,
+        if (data != null) 'data': base64Encode(data!),
       };
 
   @override