Networking de bajo nivel en Dart con Sockets y TCP

Comunicación de red de bajo nivel con dart:io

Socket es una entidad híbrida que implementa simultáneamente las interfaces Stream<Uint8List> e IOSink. Esta arquitectura permite que un socket de TCP actúe como el receptor de datos (flujo asíncrono de bytes provenientes del kernel) y como el emisor (sumidero de bytes hacia la pila de red) de forma integrada. En el fondo, dart:io utiliza las llamadas al sistema no bloqueantes (como epoll en Linux o kqueue en macOS) para notificar al event loop de la VM cuando hay datos listos para ser procesados, evitando que la ejecución se detenga mientras se espera la respuesta de la red.

Cuando implementas un protocolo binario personalizado (como MQTT o un protocolo de videojuegos), necesitas este control sobre el flujo de bytes sin la sobrecarga de abstracciones de alto nivel. Sin embargo, si intentas saturar un Socket enviando datos mediante add() más rápido de lo que el cliente puede consumirlos, provocarás un problema de backpressure (contrapresión), donde la memoria de tu proceso crecerá indefinidamente para almacenar los datos en el buffer interno. Para evitarlo, debes usar await socket.flush() o controlar el flujo mediante la pausa del stream. Si necesitas el máximo rendimiento posible, eliminando incluso el buffering intermedio que gestiona Socket, puedes recurrir a RawSocket, que te entrega los bytes directamente desde el descriptor de archivo del sistema operativo.

Para servidores, ServerSocket.bind() es el punto de partida que escucha en una interfaz específica (usando InternetAddress) y emite nuevos Socket cada vez que un cliente establece una conexión. Si necesitas seguridad, SecureSocket envuelve estas operaciones con una capa TLS/SSL mediante un SecurityContext. Para comunicaciones sin conexión (UDP), el modelo cambia radicalmente: se utiliza RawDatagramSocket para manejar paquetes independientes e inconexos.

import 'dart:io';
import 'dart:typed_data';

void main() async {
  // 1. Levantamos un servidor TCP en la interfaz local
  final server = await ServerSocket.bind(InternetAddress.loopbackIPv4, 8080);
  print('Servidor escuchando en ${server.address.address}:${server.port}');

  // Simulamos un cliente que se conecta inmediatamente
  _runClient();

  // 2. Manejo de conexiones entrantes
  await for (final Socket client in server) {
    _handleClient(client);
  }
}

Future<void> _handleClient(Socket client) async {
  final remote = client.remoteAddress;
  print('Nueva conexión desde ${remote.address}:${client.getPort()}');

  // El Socket actúa como Stream<List<int>> para recibir datos
  // Usamos listen para procesar los bytes conforme llegan al buffer de la VM
  await for (final Uint8List data in client) {
    print('Recibido: ${data.length} bytes -> ${String.fromCharCodes(data)}');

    // El Socket actúa como IOSink para enviar datos
    // Escribimos un eco del mensaje recibido
    client.add('Echo: '.bytes);
    client.add(data);
    
    // flush() es crítico: asegura que los bytes salgan del buffer 
    // de la aplicación antes de continuar la lógica.
    await client.flush();
  }

  print('Cliente desconectado: ${client.remoteAddress.address}');
  await client.close();
}

Future<void> _runClient() async {
  // Pequeño delay para asegurar que el servidor esté listo
  await Future.delayed(const Duration(seconds: 1));

  try {
    final socket = await Socket.connect('localhost', 8080);
    print('Cliente conectado a localhost:8080');

    socket.write('Hola, Dart VM!');
    
    // Escuchar la respuesta del servidor (el eco)
    await for (final Uint8List response in socket) {
      print('Cliente recibió: ${String.fromCharCodes(response)}');
      if (String.fromCharCodes(response).contains('Eco')) {
        await socket.close();
        return;
      }
    }
  } catch (e) {
    print('Error en el cliente: $e');
  }
}

extension on String {
  Uint8List get bytes => Uint8List.fromList(this.codeUnits);
}

En el ejemplo anterior, ServerSocket.bind reserva el puerto en el sistema operativo y bloquea el hilo de ejecución de forma asíncrona hasta que la escucha es exitosa. Cuando _runClient establece la conexión, el loop de eventos de la VM detecta la señal de conexión y activa el await for en _handleClient.

Es vital observar cómo client se itera directamente: esto es posible porque Socket implementa Stream<List<int>>. Cada vez que el kernel recibe un paquete TCP y lo reensambla, el Stream emite un Uint8List. Dentro del bucle, usamos client.add() para enviar datos. Es imperativo notar el uso de await client.flush(); sin esto, si intentas cerrar el socket inmediatamente después de un add(), es probable que el buffer interno no se haya vaciado y los datos se pierdan, ya que add() es una operación asíncrona que solo añade datos a una cola interna.

El error frecuente

Un error común al trabajar con Stream en sockets es ignorar los eventos de error dentro del flujo de datos. Si un cliente cierra la conexión abruptamente o hay un error de red, el await for puede fallar silenciosamente o lanzar una excepción no controlada que mata el proceso si no se gestiona.

// ERROR: No hay manejo de errores en el stream
await for (final data in client) {
  // Si el cliente cierra la conexión de forma errática, 
  // esto lanzará una excepción que no está capturada.
  process(data);
}

// CORRECTO: Escuchar explícitamente los errores del socket
client.listen(
  (data) => process(data),
  onError: (error) => print('Error en el flujo: $error'),
  onDone: () => print('Conexión cerrada'),
  cancelOnError: true,
);

Además, un error de arquitectura es realizar procesamiento pesado de los bytes recibidos (como parsear un JSON gigante o descifrar un payload) directamente dentro del listen del socket. Esto bloqueará el event loop y el servidor dejará de aceptar nuevas conexiones o de leer de otros sockets. En esos casos, debes delegar la computación a un Isolate separado.

N° 77