stream_test_6.dart

import 'dart:async';

void main() async {
  /// assume a simplified game message using a record type: (int gameId, String eventData)
  var streamController = StreamController<(int, String)>();
  var stream = streamController.stream;
  var messageSwitch = Switch(stream);

  // game 1, event 1a
  streamController.sink.add((1, '1a'));
  // game 2, event 2a
  streamController.sink.add((2, '2a'));
  // and so on...
  streamController.sink.add((1, '1b'));
  streamController.sink.add((2, '2b'));
}

class Switch {
  final Stream<(int, String)> stream;
  late StreamSubscription _subscription;
  final Map<int, GameStateController> _gameStateControllers = {};

  Switch(Stream<(int, String)> stream) : stream = stream.asBroadcastStream() {
    _subscription = this.stream.listen(eventListener);
  }

  void eventListener((int, String) event) {
    print('Switch: received event $event');
    var gameId = event.$1;
    var gameStateController = _gameStateControllers.putIfAbsent(gameId, () {
      var streamController = StreamController<(int, String)>();
      var gameEventStream = streamController.stream;
      streamController.sink.add(event);

      // adding the broadcast stream's filtered events to the game's stream
      var parentStream = stream.where((event) => gameId == event.$1);
      unawaited(streamController.addStream(parentStream));

      var gameStateController = GameStateController(gameId, gameEventStream);
      gameStateController.initialize().then((value) {
        unawaited(gameStateController.startEventProcessing());
      });
      return gameStateController;
    });
  }
}

class GameStateController {
  final int gameId;
  final Stream<(int, String)> _stream;

  GameStateController(this.gameId, Stream<(int, String)> eventStream) : _stream = eventStream;

  Future<void> initialize() async {
    print('GameStateController$gameId: initializing...');
    // delay 100ms
    await Future.delayed(Duration(milliseconds: 100));
    print('GameStateController$gameId: initialized');
  }

  Future<void> startEventProcessing() async {
    await for (var event in _stream) {
      await processEvent(event);
    }
  }

  Future<void> processEvent(event) async {
    print('GameStateController$gameId: received event $event');
    // delay 50ms
    await Future.delayed(Duration(milliseconds: 50));
    print('GameStateController$gameId: processed event $event');
  }
}