Flutter中实现TCP通信的关键步骤与代码示例

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

class TcpClient {
  final String host;
  final int port;
  Socket? _socket;
  StreamSubscription<String>? _subscription;

  TcpClient(this.host, this.port);

  Future<void> connect() async {
    try {
      _socket = await Socket.connect(host, port, timeout: const Duration(seconds: 5));
      print('✅ Connected to: ${_socket!.remoteAddress.address}:${_socket!.remotePort}');

      _subscription = _socket!
          .transform(utf8.decoder)
          .transform(const LineSplitter())
          .listen(_onLine, onError: _onError, onDone: _onDone);
    } catch (e) {
      print('🚫 Connection failed: $e');
      rethrow;
    }
  }

  void _onLine(String line) {
    print('📩 Received line: $line');
  }

  void _onError(Object e, [StackTrace? st]) {
    print('❌ Socket error: $e');
    disconnect();
  }

  void _onDone() {
    print('⚠️ Server closed connection');
    disconnect();
  }

  void send(String message) {
    final s = _socket;
    if (s != null) {
      s.write(message + '\n'); // 每条消息后加换行
      print('📤 Sent: $message');
    }
  }

  void disconnect() {
    _subscription?.cancel();
    _subscription = null;
    _socket?.destroy();
    _socket = null;
    print('🔌 Disconnected');
  }
}

class HeartbeatManager {
  final void Function() onSendHeartbeat;
  final Duration interval;
  Timer? _timer;

  HeartbeatManager({required this.onSendHeartbeat, this.interval = const Duration(seconds: 30)});

  void start() {
    _timer ??= Timer.periodic(interval, (_) => onSendHeartbeat());
  }

  void stop() {
    _timer?.cancel();
    _timer = null;
  }
}

import 'dart:math';

class ReconnectPolicy {
  final Duration minBackoff;
  final Duration maxBackoff;
  int _attempt = 0;
  final Random _rnd = Random();

  ReconnectPolicy({this.minBackoff = const Duration(seconds: 1), this.maxBackoff = const Duration(seconds: 30)});

  Duration nextDelay() {
    final base = minBackoff.inMilliseconds * pow(2, _attempt).toInt();
    final capped = min(base, maxBackoff.inMilliseconds);
    final jitter = (capped * (0.2 * (_rnd.nextDouble() * 2 - 1))).round();
    _attempt = min(_attempt + 1, 10);
    return Duration(milliseconds: max(0, capped + jitter));
  }

  void reset() => _attempt = 0;
}

import 'dart:async';
import 'dart:convert';
import 'dart:io';
import 'dart:math';

class ReconnectPolicy {
  final Duration minBackoff;
  final Duration maxBackoff;
  int _attempt = 0;
  final Random _rnd = Random();

  ReconnectPolicy({this.minBackoff = const Duration(seconds: 1), this.maxBackoff = const Duration(seconds: 30)});

  Duration nextDelay() {
    final base = minBackoff.inMilliseconds * pow(2, _attempt).toInt();
    final capped = min(base, maxBackoff.inMilliseconds);
    final jitter = (capped * (0.2 * (_rnd.nextDouble() * 2 - 1))).round();
    _attempt = min(_attempt + 1, 10);
    return Duration(milliseconds: max(0, capped + jitter));
  }

  void reset() => _attempt = 0;
}

class HeartbeatManager {
  final void Function() onSendHeartbeat;
  final Duration interval;
  Timer? _timer;

  HeartbeatManager({required this.onSendHeartbeat, this.interval = const Duration(seconds: 30)});

  void start() {
    _timer ??= Timer.periodic(interval, (_) => onSendHeartbeat());
  }

  void stop() {
    _timer?.cancel();
    _timer = null;
  }
}

class RobustLineClient {
  final String host;
  final int port;
  Socket? _socket;
  StreamSubscription<String>? _sub;

  final HeartbeatManager _hb;
  final ReconnectPolicy _policy = ReconnectPolicy();
  Timer? _idleTimer;

  RobustLineClient({required this.host, required this.port})
      : _hb = HeartbeatManager(onSendHeartbeat: () {/* later bound */}, interval: const Duration(seconds: 20));

  Future<void> start() async {
    await _connect();
  }

  Future<void> _connect() async {
    try {
      _socket = await Socket.connect(host, port, timeout: const Duration(seconds: 5));
      print('✅ connected');
      _policy.reset();
      _hb.start();

      _sub = _socket!
          .transform(utf8.decoder)
          .transform(const LineSplitter())
          .listen(_onLine, onError: _onError, onDone: _onDone);

      // 心跳绑定到 send
      _hb.onSendHeartbeat.call = () => send('ping');

      _resetIdleTimeout();
    } catch (e) {
      print('🚫 connect failed: $e');
      await _scheduleReconnect();
    }
  }

  void _onLine(String line) {
    _resetIdleTimeout();
    print('📩 $line');
  }

  void _onError(Object e, [StackTrace? st]) {
    print('❌ $e');
    _teardown();
    _scheduleReconnect();
  }

  void _onDone() {
    print('⚠️ closed by server');
    _teardown();
    _scheduleReconnect();
  }

  void _teardown() {
    _idleTimer?.cancel();
    _idleTimer = null;
    _hb.stop();
    _sub?.cancel();
    _sub = null;
    _socket?.destroy();
    _socket = null;
  }

  Future<void> _scheduleReconnect() async {
    final delay = _policy.nextDelay();
    print('⏳ reconnect in ${delay.inMilliseconds} ms');
    await Future.delayed(delay);
    await _connect();
  }

  void _resetIdleTimeout() {
    _idleTimer?.cancel();
    _idleTimer = Timer(const Duration(seconds: 60), () {
      print('⏰ idle timeout -> reconnect');
      _teardown();
      _scheduleReconnect();
    });
  }

  void send(String message) {
    _socket?.write(message + '\n');
  }
}