Dart How to implements a continuous output web-socket

Question

How to realize the continuous output of data from the server to the client. For example, after receiving the start command from the client, the server keeps outputting the current time to the client until the server receives the quit command.

Questions:

1. Why can't the server receive the quit command
2. Why only the first client can receive the continuous output message from the server

My code in Dart is

Server:

import 'dart:io';

main() async {
  var running = true;
  var q = 'quit';
  var p;
  final port = 8080;
  
  var webSocketTransformer = WebSocketTransformer();
  
  HttpServer server = await HttpServer.bind(InternetAddress.anyIPv6, port);
  server.transform(webSocketTransformer).listen((WebSocket webSocket) {
    print('WebSocket opened.');

    webSocket.listen((_) {
      q = _;
      print(_);
    }, onError: (err) {
      running = false;
      print(err);
    }, onDone: () {
      running = false;
      print('WebSocket closed.');
    });

    while (running && p != q) {
      sleep(Duration(seconds: 1));
      webSocket.add(DateTime.now().toString());
    }
  });
  print('Listening..');
}

Client:

import 'dart:io';
import 'package:web_socket_channel/io.dart';

void main(List<String> arguments) {
  final url = 'ws://localhost:8080';
  final channel = IOWebSocketChannel.connect(url);

  channel.sink.add('start');
  channel.stream.listen((msg) {
    print(msg);
  });

  sleep(Duration(seconds: 30));
  channel.sink.add('quit');
}

Answer 1

I think your current approach will block the socket from further clients if it's busy in an infinite loop. Your best bet is to use a Periodic Stream that will emit messages at a given interval so it can still listen for future clients.

Server:

import 'dart:io';
import 'dart:async';

main() async {
  final port = 8080;
  
  var webSocketTransformer = WebSocketTransformer();

  // Use a periodic stream that emits events at a given interval
  Stream<int> streamPeriodic =
          Stream.periodic(const Duration(milliseconds: 1000), (count) {
        return count;
   });
  late StreamSubscription<int> subscription;
  bool openSocket = false;

  HttpServer server = await HttpServer.bind(InternetAddress.anyIPv6, port);
  server.transform(webSocketTransformer).listen((WebSocket webSocket) {
       print('WebSocket opened.');
       
       webSocket.listen( (event) {
         if (!openSocket) {
              subscription = streamPeriodic.listen((event) {
                   webSocket.add(DateTime.now().toString());
              });
            openSocket = true;
         }
         else {
             event as String;
             if (event == 'start') {
                subscription.resume();
             }
             else if (event == 'quit') {
                subscription.pause();  
             }
         }
    });         
  });
  print('Listening..');
}