SocketException: Connection reset by peer (OS Error: Connection reset by peer, errno = 104) about mqtt_client HOT 10 CLOSED

class MqttClient extends ClientHandler {
  MqttServerClient? _client;
  String? _clientId;
  StreamController<PayloadWithTopic>? _messagesController =
      StreamController.broadcast();
  BehaviorSubject<ConnectionState> _cnxBehavior =
      BehaviorSubject<ConnectionState>();
  int maxReconnect = 10;
  int _crtReconnect = 0;
  List<String> _subscribedTopics = [];

  @override
  Future<bool> connect(
      {required host,
      required String username,
      required String password,
      String? clientId,
      int? port}) async {
    String cid = clientId ?? getClientId()!;
    _clientId = cid;
    _client =
        MqttServerClient.withPort(host, cid, 1883, maxConnectionAttempts: 3);
    debugPrint(
        '!!!! ------------> [conn] connect with $host, $port, $username, $password, ${_clientId}');

    // _client!.logging(on: true);
    _client!.logging(on: false);
    _client!.onConnected = onConnected;
    _client!.onDisconnected = onDisconnected;
    _client!.onUnsubscribed = onUnsubscribed;
    _client!.onSubscribed = onSubscribed;
    _client!.onSubscribeFail = onSubscribeFail;
    _client!.onAutoReconnect = onAutoReconnect;
    _client!.pongCallback = pong;
    _client!.keepAlivePeriod = 60;
    _client!.autoReconnect = false;

    final connMessage = MqttConnectMessage()
        .authenticateAs(username, password)
        .withWillTopic('lastwills')
        .withWillMessage('Will message')
        .withClientIdentifier(cid)
        .withProtocolName("MQTT")
        .withProtocolVersion(4)
        .withWillQos(MqttQos.atLeastOnce);
    _client!.connectionMessage = connMessage;
    try {
      //_messagesController =  StreamController.broadcast();
      _broadcastConnectionState();
      await _client!.connect();

      _broadcastConnectionState();
      _listenAndFilter();
      return true;
    } catch (e) {
      kLog('Exception: $e');
      if (_client != null) _client!.disconnect();
      return false;
    }
  }

  void _subscribeToArchivesTopics() {
    _client!
        .subscribe(getClientId()!.toArchivesRoomsTopic, MqttQos.atLeastOnce);
    _client!
        .subscribe(getClientId()!.toArchivesMessagesTopic, MqttQos.atLeastOnce);
    _client!.subscribe(getClientId()!.toArchivesMyId, MqttQos.atLeastOnce);
    debugPrint("[client subscribe archives] just after connect");
  }

  void _listenAndFilter() {
    _client!.updates!.listen((List<MqttReceivedMessage<MqttMessage>> c) {
      final MqttPublishMessage message = c[0].payload as MqttPublishMessage;
      String topic = c[0].topic;
      // final payload =
      //     MqttPublishPayload.bytesToStringAsString(message.payload.message);
      final payload = Utf8Decoder().convert(message.payload.message);
      debugPrint('[_listenAndFilter] Got msg: $topic, $payload');
      // var a = utf8.decode(payload.codeUnits);
      // print(payload);
      _messagesController!
          .add(PayloadWithTopic(payload: payload, topic: topic));
      debugPrint(
          '[_listenAndFilter] Received message:$payload from topic: ${c[0].topic}');
    });
  }

  _broadcastConnectionState() {
    if (_client == null) {
      debugPrint('[broadcast status] client is null, disconnected...');
      _cnxBehavior.add(ConnectionState.disconnected);
      connectState.value = ConnectionState.disconnected;
      return;
    }
    if (_client!.connectionStatus == null) {
      print('[broadcast status] connectionStatus is null, disconnected...');
      _cnxBehavior.add(ConnectionState.disconnected);
      connectState.value = ConnectionState.disconnected;
      return;
    }
    switch (_client!.connectionStatus!.state) {
      case MqttConnectionState.disconnecting:
        _cnxBehavior.add(ConnectionState.disconnecting);
        connectState.value = ConnectionState.disconnecting;
        break;
      case MqttConnectionState.disconnected:
        _cnxBehavior.add(ConnectionState.disconnected);
        connectState.value = ConnectionState.disconnected;
        break;
      case MqttConnectionState.connecting:
        _cnxBehavior.add(ConnectionState.connecting);
        connectState.value = ConnectionState.connecting;
        break;
      case MqttConnectionState.connected:
        _cnxBehavior.add(ConnectionState.connected);
        connectState.value = ConnectionState.connected;
        break;
      case MqttConnectionState.faulted:
        _cnxBehavior.add(ConnectionState.faulted);
        connectState.value = ConnectionState.faulted;
        break;
    }
  }

  @override
  bool isConnected() {
    if (_client == null) {
      debugPrint("[isConnected] _client is null!! not connected");
      return false;
    }
    if (_client!.connectionStatus == null) {
      debugPrint("[isConnected] _client connectionStatus null: not connected");
      return false;
    }
    return _client!.connectionStatus!.state == MqttConnectionState.connected;
  }

  @override
  bool isConnecting() {
    if (_client != null && _client!.connectionStatus != null) {
      return _client!.connectionStatus!.state == MqttConnectionState.connecting;
    }
    return false;
  }

  @override
  void joinRoom(String bareRoom) {
    String messagesTopic = bareRoom.toChattingTopic;
    String eventsTopic = bareRoom.toRoomEventsTopic;
    subscribeTopicOnlyOnce(messagesTopic);
    subscribeTopicOnlyOnce(eventsTopic);
  }

  @override
  void leaveRoom(String bareRoom) {
    String messagesTopic = bareRoom.toChattingTopic;
    String eventsTopic = bareRoom.toRoomEventsTopic;
    if (_client == null) {
      return;
    }
    _client!.unsubscribe(messagesTopic);
    _client!.unsubscribe(eventsTopic);
    _subscribedTopics.remove(messagesTopic);
    _subscribedTopics.remove(eventsTopic);
  }

  @override
  void joinContactEvents(String contactId) {
    subscribeTopicOnlyOnce(contactId.toPresenceTopic);
  }

  void subscribeTopicOnlyOnce(String topic) {
    if (!_subscribedTopics.contains(topic)) {
      _client!.subscribe(topic, MqttQos.atLeastOnce);
      _subscribedTopics.add(topic);
      debugPrint("[JION ROOM] !!!! ******** join roomId: $topic");
    }
  }

  @override
  void joinMyEvents(String myId) {
    if (isConnected()) {
      subscribeTopicOnlyOnce(myId.toPersonalEventTopic);
      subscribeTopicOnlyOnce(myId.toArchivesMessagesTopic);
    } else {
      debugPrint("client not connected, pass joinMyEvents");
    }
  }

  @override
  void leaveContactEvents(String contactId) {
    if (_client == null) {
      return;
    }
    _client!.unsubscribe(contactId.toPresenceTopic);
  }

  @override
  void sendPayload(String payload, String channel) {
    final builder = MqttClientPayloadBuilder();
    builder.addUTF8String(payload);
    debugPrint("sending payload client? $_client");
    if (_client != null) {
      _client!.publishMessage(channel, MqttQos.atLeastOnce, builder.payload!);
    } else {
      // do with disconnected
    }
  }

  @override
  void sendFilePayload(File file, String channel) {
    final bytes = file.readAsBytesSync();
    var mime = lookupMimeType(file.path);
    String base64Image = "data:" +
        (mime ?? "text/plain") +
        ";base64," +
        base64Encode(bytes) +
        "," +
        basename(file.path);
    debugPrint('[SENDING... FILE] $base64Image, $channel');
    sendPayload(base64Image, channel);
  }

  @override
  void disconnect() {
    try {
      _client!.disconnect();
    } catch (e) {
      kLog("[disconnect error] $e");
    }
    _client = null;
    _clientId = null;
  }

  @override
  Stream<PayloadWithTopic> allMessagesStream() {
    return _messagesController!.stream;
  }

// connection succeeded
  void onConnected() {
    debugPrint('[onConnected] Connected...');
    _broadcastConnectionState();
      _subscribeToArchivesTopics();

  }

// unconnected
  void onDisconnected() {
    debugPrint('[Disconnected] Disconnected...');
    _broadcastConnectionState();
    if (_messagesController != null && !_messagesController!.isClosed) {
      //_messagesController!.close();
    }
    _client = null;
    _clientId = null;
  }

// subscribe to topic succeeded
  void onSubscribed(String topic) {
    debugPrint('Subscribed topic: $topic');
  }

// subscribe to topic failed
  void onSubscribeFail(String topic) {
    debugPrint('Failed to subscribe $topic');
  }

// unsubscribe succeeded
  void onUnsubscribed(String? topic) {
    debugPrint('Unsubscribed topic: $topic');
  }

}