Added option for coordinator to participate in UserCoordinationMessages (via inlet).

Author: zeyus

packages/liblsl_coordinator/example/multi_node_test.dart

@@ -91,6 +91,7 @@ Future<void> _runNode({
       discoveryInterval: Duration(seconds: 5),
       nodeTimeout: Duration(seconds: 10),
       maxNodes: maxNodes, // This may cause some nodes to be rejected
+      consumeCoordinationStreamAsCoordinator: false,
     );
 
     // Create coordination configuration

packages/liblsl_coordinator/lib/src/network/coordination_session.dart

@@ -34,13 +34,24 @@ class CoordinationSessionConfig implements IConfig {
   /// This should be at least twice the [heartbeatInterval].
   final Duration nodeTimeout;
 
+  /// Determines whether on becoming coordinator, the node should
+  /// also consume the coordination stream - specifically, this is used for
+  /// a coordinator to be able to consume UserMessages (UserCoordinationMessage)
+  /// within the coordination session, this allows for a session coordinator
+  /// to also act as a participant in the session, and have similar latency
+  /// characteristics as other nodes.
+  /// Default is true.
+
+  final bool consumeCoordinationStreamAsCoordinator;
+
   CoordinationSessionConfig({
     required this.name,
     this.maxNodes = 10,
     this.minNodes = 1,
     this.heartbeatInterval = const Duration(seconds: 5),
     this.discoveryInterval = const Duration(seconds: 10),
     this.nodeTimeout = const Duration(seconds: 15),
+    this.consumeCoordinationStreamAsCoordinator = true,
   }) {
     validate(throwOnError: true);
   }
@@ -100,14 +111,18 @@ class CoordinationSessionConfig implements IConfig {
       'heartbeatInterval': heartbeatInterval.inMilliseconds,
       'discoveryInterval': discoveryInterval.inMilliseconds,
       'nodeTimeout': nodeTimeout.inMilliseconds,
+      'consumeCoordinationStreamAsCoordinator':
+          consumeCoordinationStreamAsCoordinator,
     };
   }
 
   @override
   String toString() {
     return 'NetworkSessionConfig(name: $name, maxNodes: $maxNodes, '
         'minNodes: $minNodes, heartbeatInterval: $heartbeatInterval, '
-        'discoveryInterval: $discoveryInterval, nodeTimeout: $nodeTimeout)';
+        'discoveryInterval: $discoveryInterval, nodeTimeout: $nodeTimeout, '
+        'consumeCoordinationStreamAsCoordinator: '
+        '$consumeCoordinationStreamAsCoordinator)';
   }
 
   @override
@@ -118,6 +133,7 @@ class CoordinationSessionConfig implements IConfig {
     Duration? heartbeatInterval,
     Duration? discoveryInterval,
     Duration? nodeTimeout,
+    bool? consumeCoordinationStreamAsCoordinator,
   }) {
     return CoordinationSessionConfig(
       name: name ?? this.name,
@@ -126,6 +142,9 @@ class CoordinationSessionConfig implements IConfig {
       heartbeatInterval: heartbeatInterval ?? this.heartbeatInterval,
       discoveryInterval: discoveryInterval ?? this.discoveryInterval,
       nodeTimeout: nodeTimeout ?? this.nodeTimeout,
+      consumeCoordinationStreamAsCoordinator:
+          consumeCoordinationStreamAsCoordinator ??
+          this.consumeCoordinationStreamAsCoordinator,
     );
   }
 
@@ -137,6 +156,7 @@ class CoordinationSessionConfig implements IConfig {
       heartbeatInterval: const Duration(seconds: 5),
       discoveryInterval: const Duration(seconds: 10),
       nodeTimeout: const Duration(seconds: 15),
+      consumeCoordinationStreamAsCoordinator: true,
     );
   }
 
@@ -151,7 +171,9 @@ class CoordinationSessionConfig implements IConfig {
         other.minNodes == minNodes &&
         other.heartbeatInterval == heartbeatInterval &&
         other.discoveryInterval == discoveryInterval &&
-        other.nodeTimeout == nodeTimeout;
+        other.nodeTimeout == nodeTimeout &&
+        other.consumeCoordinationStreamAsCoordinator ==
+            consumeCoordinationStreamAsCoordinator;
   }
 
   @override
@@ -162,7 +184,8 @@ class CoordinationSessionConfig implements IConfig {
         minNodes.hashCode ^
         heartbeatInterval.hashCode ^
         discoveryInterval.hashCode ^
-        nodeTimeout.hashCode;
+        nodeTimeout.hashCode ^
+        consumeCoordinationStreamAsCoordinator.hashCode;
   }
 }
 

packages/liblsl_coordinator/lib/transports/lsl/lsl_coordination_controller.dart

@@ -233,6 +233,14 @@ class CoordinationController {
     _state.transitionTo(CoordinationPhase.accepting);
 
     logger.fine('Coordinator ready, accepting nodes');
+    if (coordinationConfig
+        .sessionConfig
+        .consumeCoordinationStreamAsCoordinator) {
+      // Connect to own coordinator stream as participant
+      await _connectToCoordinator(thisNode.uId);
+    } else {
+      logger.info('Not consuming own coordinator stream as per configuration');
+    }
   }
 
   /// Become a participant
@@ -304,11 +312,23 @@ class CoordinationController {
     logger.finer(
       '[CONTROLLER-${thisNode.uId}] Found coordinator stream, adding inlet...',
     );
-    // parse streaminfo
-    final info = LSLStreamInfoHelper.parseSourceId(streamInfos.first.sourceId);
-    final nodeUId = info[LSLStreamInfoHelper.nodeUIdKey]!;
-    // set coordinator UId in state
-    _state.becomeParticipant(nodeUId);
+    if (thisNode.uId != coordinatorUId) {
+      // parse streaminfo
+      final info = LSLStreamInfoHelper.parseSourceId(
+        streamInfos.first.sourceId,
+      );
+      final nodeUId = info[LSLStreamInfoHelper.nodeUIdKey]!;
+      logger.fine(
+        '[CONTROLLER-${thisNode.uId}] Connecting to coordinator stream of node $nodeUId as participant',
+      );
+      // set coordinator UId in state
+      _state.becomeParticipant(nodeUId);
+    } else {
+      logger.fine(
+        '[CONTROLLER-${thisNode.uId}] Connected to own coordinator stream (self-coordination)',
+      );
+    }
+
     await _coordinationStream.addInlet(streamInfos.first);
     logger.info(
       '[CONTROLLER-${thisNode.uId}] Connected to coordinator stream successfully',

packages/liblsl_coordinator/pubspec.yaml

@@ -1,6 +1,6 @@
 name: liblsl_coordinator
 description: A performance-focused Dart (and Flutter) LSL-based device coordination library.
-version: 0.0.5+0
+version: 0.0.6+0
 homepage: https://zeyus.com/
 repository: https://github.com/NexusDynamic/liblsl.dart
 issue_tracker: https://github.com/NexusDynamic/liblsl.dart/issues

packages/liblsl_test/pubspec.yaml

@@ -22,7 +22,7 @@ resolution: workspace
 # https://developer.apple.com/library/archive/documentation/General/Reference/InfoPlistKeyReference/Articles/CoreFoundationKeys.html
 # In Windows, build-name is used as the major, minor, and patch parts
 # of the product and file versions while build-number is used as the build suffix.
-version: 1.1.0+1
+version: 1.1.1+1
 
 environment:
   sdk: ^3.9.0