connect method

@override Future<MqttClientConnectionStatus?> connect(String server, int port)

Connect

Implementation

@override
Future<MqttClientConnectionStatus?> connect(String server, int port) {
  final completer = Completer<MqttClientConnectionStatus?>();
  MqttLogger.log('MqttWsConnection::connect - entered');
  // Add the port if present
  Uri uri;
  try {
    uri = Uri.parse(server);
  } on Exception catch (_, stack) {
    final message =
        'MqttWsConnection::connect - The URI supplied for the WS '
        'connection is not valid - $server';
    Error.throwWithStackTrace(NoConnectionException(message), stack);
  }
  if (uri.scheme != 'ws' && uri.scheme != 'wss') {
    final message =
        'MqttWsConnection::connect - The URI supplied for the WS has '
        'an incorrect scheme - $server';
    throw NoConnectionException(message);
  }

  uri = uri.replace(port: port);

  final uriString = uri.toString();
  MqttLogger.log(
    'MqttWsConnection::connect - WS URL is $uriString, protocols are $protocols',
  );
  HttpClient? httpClient;
  if (onBadCertificate != null) {
    httpClient = HttpClient()
      ..badCertificateCallback = (cert, host, port) {
        return onBadCertificate!(cert);
      };
  }
  // The TCP connect is also given the connection timeout, so the runtime
  // cancels a connect still pending when it elapses.
  if (connectionTimeout != null) {
    httpClient ??= HttpClient();
    httpClient.connectionTimeout = connectionTimeout;
  }
  try {
    // Connect and save the socket.
    _webSocketWithinConnectionTimeout(
          WebSocket.connect(
            uriString,
            protocols: protocols.isNotEmpty ? protocols : null,
            headers: headers,
            customClient: httpClient,
          ),
          uriString,
        )
        .then((socket) {
          client = socket;
          readWrapper = ReadWrapper();
          messageStream = MqttByteBuffer(typed.Uint8Buffer());
          _startListening();
          completer.complete();
        })
        .catchError((e) {
          onError(e);
          completer.completeError(e);
        });
  } on Exception catch (_, stack) {
    final message =
        'MqttWsConnection::connect - The connection to the message broker '
        '{$uriString} could not be made.';
    Error.throwWithStackTrace(NoConnectionException(message), stack);
  }
  return completer.future;
}