Batching

example/console/.gitignore

@@ -0,0 +1,2 @@
+node_modules
+package-lock.json

example/console/Dockerfile

@@ -0,0 +1,5 @@
+FROM node:14.15.1-alpine3.12
+WORKDIR /app
+COPY package.json .
+RUN npm i --quiet
+COPY sign-server.js ./

example/console/config.json

@@ -0,0 +1,27 @@
+{
+  "secret": "secret",
+  "api_key": "strong_secret_for_api_auth",
+  "admin": true,
+  "admin_password": "password",
+  "admin_secret": "strong_secret_key_to_sign_authorization_token",
+  "internal_port": "10001",
+  "max_channel_length": 500,
+  "connection_lifetime": 60,
+  "log_level": "debug",
+  "namespaces": [
+    {
+      "name": "public",
+      "publish": true,
+      "anonymous": true,
+      "presence": true,
+      "join_leave": false
+    },
+    {
+      "name": "user",
+      "anonymous": false,
+      "publish": true,
+      "presence": false,
+      "join_leave": false
+    }
+  ]
+}

example/console/docker-compose.yml

@@ -0,0 +1,22 @@
+version: "3"
+services:
+  centrifugo:
+    image: centrifugo/centrifugo:v2.3.1
+    container_name: centrifugo
+    restart: always
+    ports:
+      - 8000:8000
+    volumes:
+      - ./:/centrifugo
+    command: centrifugo --config=config.json
+  sign_server:
+    container_name: sign_server
+    build: .
+    command: npm start
+    volumes:
+      - ./:/app
+      - /app/node_modules
+    ports:
+      - 5000:5000
+    environment:
+      - PORT=5000

example/console/package.json

@@ -0,0 +1,17 @@
+{
+  "name": "console",
+  "version": "1.0.0",
+  "description": "Before running example make sure you created `chat` namespace in Centrifugo configuration and allowed publishing into channel - i.e. using config like this:",
+  "main": "sign-server.js",
+  "scripts": {
+    "start": "node sign-server.js"
+  },
+  "keywords": [],
+  "author": "",
+  "license": "ISC",
+  "dependencies": {
+    "body-parser": "^1.19.0",
+    "express": "^4.17.1",
+    "jsonwebtoken": "^8.5.1"
+  }
+}

example/console/readme.md

@@ -1,23 +1,23 @@
-Before running example make sure you created `chat` namespace in Centrifugo configuration and allowed publishing into channel - i.e. using config like this:
+# centrifuge-dart console app
 
-```json
-{
-  ...
-  "namespaces": [
-    {
-      "name": "chat",
-      "anonymous": false,
-      "publish": true,
-      "join_leave": true,
-      "presence": true,
-      "presence_stats": true
-    }
-  ]
-}
-```
+You can run environment in two ways:
 
-Also run Centrifugo in insecure client mode so it does not expect JWT token from client:
+1. Docker (recommended)
+   ```bash
+   docker-compose up [-d]
+   ```
 
+2. Manually
+   ```bash
+   # first terminal session
+   npm i
+   npm start 
+
+   # second terminal session
+   ./centrifugo --config config.json --client_insecure
+   ```
+
+To run console app itself:
 ```bash
-./centrifugo --config config.json --client_insecure
-```
+dart ./simple.dart
+```

example/console/sign-server.js

@@ -0,0 +1,26 @@
+const express = require('express')
+const bodyParser = require('body-parser')
+const jwt = require('jsonwebtoken')
+
+const app = express()
+const PORT = process.env.PORT || 5000
+
+app.use(bodyParser.json())
+
+app.post('/auth', (req, res) => {
+  const channels = []
+
+  for (const channel of req.body.channels) {
+    const token = jwt.sign({
+      client: req.body.client,
+      channel,
+    }, 'secret')
+    channels.push({ channel, token })
+  }
+
+  res.json({ channels })
+})
+
+app.listen(PORT, () => {
+  console.log(`Sign server listening on port: ${ PORT }`)
+})

example/console/simple.dart

@@ -1,45 +1,68 @@
 import 'dart:async';
 import 'dart:convert';
 import 'dart:io';
+import 'package:http/http.dart' as http;
 
 import 'package:centrifuge/centrifuge.dart' as centrifuge;
 
 void main() async {
   final url = 'ws://localhost:8000/connection/websocket?format=protobuf';
-  final channel = 'chat:index';
+  final channel = 'public:test';
 
-  final onEvent = (dynamic event) {
+  // Uncomment to subscribe to private channels
+  // final channel = r'$user:test';
+
+  // Uncomment to use batching
+  // final channels = [
+  //   r'$usert:test1',
+  //   r'$user:test2',
+  //   r'$user:test3',
+  //   r'public:test1',
+  // ];
+
+  final onEvent = (String channel, dynamic event) {
     print('$channel> $event');
   };
 
   try {
+    final httpClient = http.Client();
     final client = centrifuge.createClient(
       url,
       config: centrifuge.ClientConfig(
-          headers: <String, dynamic>{'user-id': 42, 'user-name': 'The Answer'},
-          onPrivateSub: (centrifuge.PrivateSubEvent event) {
-            return Future.value('<SUBSCRIPTION JWT>');
-          }),
+        onPrivateSub: (event) =>
+            _auth(httpClient, event.clientID, event.channels),
+      ),
     );
 
-    client.connectStream.listen(onEvent);
-    client.disconnectStream.listen(onEvent);
+    client.connectStream.listen((e) => onEvent('', e));
+    client.disconnectStream.listen((e) => onEvent('', e));
 
     // Uncomment to use example token based on secret key `secret`.
-    // client.setToken('eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0c3VpdGVfand0In0.hPmHsVqvtY88PvK4EmJlcdwNuKFuy3BGaF7dMaKdPlw');
-    client.connect();
+    // client.setToken(
+    //     'eyJ0eXAiOiJKV1QiLCJhbGciOiJIUzI1NiJ9.eyJzdWIiOiJ0ZXN0c3VpdGVfand0In0.hPmHsVqvtY88PvK4EmJlcdwNuKFuy3BGaF7dMaKdPlw');
+    await client.connect();
 
+    // Uncomment to use batching
+    // client.startBatching();
+    // client.startSubscribeBatching();
+    // for (final channel in channels) {
     final subscription = client.getSubscription(channel);
 
-    subscription.publishStream.map((e) => utf8.decode(e.data)).listen(onEvent);
-    subscription.joinStream.listen(onEvent);
-    subscription.leaveStream.listen(onEvent);
+    subscription.publishStream
+        .map((e) => utf8.decode(e.data))
+        .listen((e) => onEvent(channel, e));
+    subscription.joinStream.listen((e) => onEvent(channel, e));
+    subscription.leaveStream.listen((e) => onEvent(channel, e));
 
-    subscription.subscribeSuccessStream.listen(onEvent);
-    subscription.subscribeErrorStream.listen(onEvent);
-    subscription.unsubscribeStream.listen(onEvent);
+    subscription.subscribeSuccessStream.listen((e) => onEvent(channel, e));
+    subscription.subscribeErrorStream.listen((e) => onEvent(channel, e));
+    subscription.unsubscribeStream.listen((e) => onEvent(channel, e));
 
     subscription.subscribe();
+    // Uncomment to use batching
+    // }
+    // client.stopSubscribeBatching();
+    // client.stopBatching();
 
     final handler = _handleUserInput(client, subscription);
 
@@ -72,7 +95,7 @@ Function(String) _handleUserInput(
         print('RPC result: ' + utf8.decode(result.data));
         break;
       case '#disconnect':
-        client.disconnect();
+        await client.disconnect();
         break;
       default:
         final output = jsonEncode({'input': message});
@@ -87,3 +110,20 @@ Function(String) _handleUserInput(
     return;
   };
 }
+
+Future<centrifuge.PrivateSubSign> _auth(
+    http.Client httpClient, String clientID, List<String> channels) async {
+  final body = json.encode(<String, dynamic>{
+    'client': clientID,
+    'channels': channels,
+  });
+  final res = await httpClient.post(
+    'http://localhost:5000/auth',
+    headers: <String, String>{
+      'content-type': 'application/json',
+      'accept': 'application/json',
+    },
+    body: body,
+  );
+  return centrifuge.PrivateSubSign.fromRawJson(res.body);
+}

lib/centrifuge.dart

@@ -2,4 +2,4 @@ export 'src/client.dart' show Client, createClient;
 export 'src/client_config.dart';
 export 'src/error.dart';
 export 'src/events.dart';
-export 'src/subscription.dart' show Subscription;
+export 'src/subscription.dart' show Subscription, PrivateSubSign; // TODO