import 'dart:async'; import 'dart:convert'; import 'package:flutter/foundation.dart'; import 'package:web_socket_channel/web_socket_channel.dart'; class BrandmeisterWebSocketClient { WebSocketChannel? _channel; StreamController>? _messageController; Timer? _heartbeatTimer; Timer? _reconnectTimer; bool _isConnecting = false; bool _shouldReconnect = true; bool _isReady = false; int _reconnectAttempts = 0; static const int _maxReconnectAttempts = 5; static const Duration _reconnectDelay = Duration(seconds: 5); final Completer _readyCompleter = Completer(); // Socket.IO Engine.IO v4 WebSocket URL static const String _wsUrl = 'wss://api.brandmeister.network/infoService/?EIO=4&transport=websocket'; Stream> get messageStream => _messageController?.stream ?? const Stream.empty(); bool get isConnected => _channel != null; bool get isReady => _isReady; Future get ready => _readyCompleter.future; Future connect() async { if (_isConnecting || isConnected) { debugPrint('WebSocket: Already connected or connecting (isConnecting: $_isConnecting, isConnected: $isConnected)'); return; } _isConnecting = true; _shouldReconnect = true; try { debugPrint('WebSocket: Connecting to $_wsUrl'); _messageController ??= StreamController>.broadcast(); _channel = WebSocketChannel.connect(Uri.parse(_wsUrl)); // Listen to incoming messages _channel!.stream.listen( _handleMessage, onError: _handleError, onDone: _handleDisconnect, cancelOnError: false, ); _reconnectAttempts = 0; _isConnecting = false; debugPrint('WebSocket: WebSocket connected, waiting for handshake (isReady: $_isReady)'); } catch (e) { _isConnecting = false; debugPrint('WebSocket: Connection error: $e'); _scheduleReconnect(); } } void _handleMessage(dynamic message) { try { final messageStr = message.toString(); debugPrint('WebSocket: Received message: $messageStr'); // Socket.IO protocol uses numbered message types // 0 = open, 2 = ping, 3 = pong, 4 = message, etc. if (messageStr.startsWith('0')) { // Open packet - connection established _handleOpenPacket(messageStr); } else if (messageStr.startsWith('2')) { // Ping packet - respond with pong _sendPong(); } else if (messageStr.startsWith('40')) { // Socket.IO namespace connection acknowledgment (must check before '4') _handleNamespaceConnect(); } else if (messageStr.startsWith('42')) { // Socket.IO event message (42 = message with data) _handleDataPacket(messageStr); } else if (messageStr.startsWith('4')) { // Other Socket.IO message types debugPrint('WebSocket: Received other type 4 message: $messageStr'); } } catch (e) { debugPrint('WebSocket: Error handling message: $e'); } } void _handleOpenPacket(String message) { try { // Parse open packet: 0{"sid":"xxx","upgrades":[],"pingInterval":25000,"pingTimeout":20000} final jsonStr = message.substring(1); final data = jsonDecode(jsonStr) as Map; final pingInterval = data['pingInterval'] as int? ?? 25000; // Set up heartbeat based on server's ping interval _heartbeatTimer?.cancel(); _heartbeatTimer = Timer.periodic( Duration(milliseconds: pingInterval), (_) => _sendPing(), ); debugPrint('WebSocket: Open packet received, ping interval: $pingInterval ms'); // Send Socket.IO namespace connection (40) _sendNamespaceConnect(); } catch (e) { debugPrint('WebSocket: Error parsing open packet: $e'); } } void _sendNamespaceConnect() { if (isConnected) { try { _channel!.sink.add('40'); debugPrint('WebSocket: Sent namespace connect (40)'); } catch (e) { debugPrint('WebSocket: Error sending namespace connect: $e'); } } } void _handleNamespaceConnect() { debugPrint('WebSocket: Namespace connected (received 40)'); // Mark connection as ready _isReady = true; if (!_readyCompleter.isCompleted) { _readyCompleter.complete(); } } void _handleDataPacket(String message) { try { // Parse data packet: 4{"event":"data","args":[...]} // or: 42["eventName",{...}] String jsonStr = message.substring(1); // Handle Engine.IO v4 format if (jsonStr.startsWith('2')) { jsonStr = jsonStr.substring(1); } final data = jsonDecode(jsonStr); Map? eventData; if (data is Map) { eventData = data; } else if (data is List && data.isNotEmpty) { // Socket.IO array format: ["eventName", {...}] or ["eventName", [...]] final eventName = data[0] as String; if (data.length > 1) { eventData = { 'event': eventName, 'data': data[1], }; } else { eventData = { 'event': eventName, }; } } if (eventData != null && _messageController != null && !_messageController!.isClosed) { _messageController!.add(eventData); } } catch (e) { debugPrint('WebSocket: Error parsing data packet: $e'); } } void sendMessage(String event, dynamic data) { if (!isConnected) { debugPrint('WebSocket: Cannot send message, not connected'); return; } if (!isReady) { debugPrint('WebSocket: Cannot send message, handshake not complete'); return; } try { // Socket.IO format: 42["eventName", data] final message = '42${jsonEncode([event, data])}'; _channel!.sink.add(message); debugPrint('WebSocket: Sent message: $message'); } catch (e) { debugPrint('WebSocket: Error sending message: $e'); } } void getDeviceStatus(String deviceId) { debugPrint('WebSocket: getDeviceStatus called for device: $deviceId (isReady: $isReady, isConnected: $isConnected)'); sendMessage('getDeviceStatus', deviceId); } void _sendPing() { if (isConnected) { try { _channel!.sink.add('2'); debugPrint('WebSocket: Sent ping'); } catch (e) { debugPrint('WebSocket: Error sending ping: $e'); } } } void _sendPong() { if (isConnected) { try { _channel!.sink.add('3'); debugPrint('WebSocket: Sent pong'); } catch (e) { debugPrint('WebSocket: Error sending pong: $e'); } } } void _handleError(Object error) { debugPrint('WebSocket: Stream error: $error'); } void _handleDisconnect() { debugPrint('WebSocket: Disconnected'); _cleanup(); if (_shouldReconnect) { _scheduleReconnect(); } } void _scheduleReconnect() { if (_reconnectAttempts >= _maxReconnectAttempts) { debugPrint('WebSocket: Max reconnect attempts reached'); return; } _reconnectAttempts++; debugPrint('WebSocket: Scheduling reconnect attempt $_reconnectAttempts in $_reconnectDelay'); _reconnectTimer?.cancel(); _reconnectTimer = Timer(_reconnectDelay, () { if (_shouldReconnect) { connect(); } }); } void _cleanup() { _heartbeatTimer?.cancel(); _heartbeatTimer = null; _channel = null; _isConnecting = false; } void disconnect() { debugPrint('WebSocket: Disconnecting'); _shouldReconnect = false; _reconnectTimer?.cancel(); _heartbeatTimer?.cancel(); try { _channel?.sink.close(); } catch (e) { debugPrint('WebSocket: Error closing channel: $e'); } _cleanup(); } void dispose() { disconnect(); _messageController?.close(); _messageController = null; } }