时间戳

首页 > 解决方案 > Flutter MQTT Client Detached socket 不能同时实现'Stream>'和'流'

flutter - Flutter MQTT Client Detached socket 不能同时实现'Stream>'和'流'

问题描述

我有一个正在运行的 mosquitto 代理在公共 ip 中运行，比如说 234.56.xx.345 在端口 1883。我正在制作一个基于颤振的 android 应用程序，以便在一段时间后分享一些信息。我包含了 mqtt 颤振包，但在编译时出现错误。下面找到我的代码。

我得到的错误是编译器消息：

file:///home/cccc/.pub-cache/hosted/pub.dartlang.org/mqtt_client-5.5.3/lib/src/connectionhandling/mqtt_client_mqtt_ws2_connection.dart:11:7: Error: '_DetachedSocket' can't implement both 'Stream<List<int>>' and 'Stream<Uint8List>'
 - 'Stream' is from 'dart:async'.
 - 'List' is from 'dart:core'.
 - 'Uint8List' is from 'dart:typed_data'.
class _DetachedSocket extends Stream<List<int>> implements Socket {
      ^
Compiler failed on /media/ccc/PRO/yebdriver/lib/main.dart
Finished with error: Gradle task assembleDebug failed with exit code 1

下面是我的mqtt客户端初始化代码：

import 'dart:async';
import 'dart:io';
import 'package:mqtt_client/mqtt_client.dart';


final MqttClient client = MqttClient('test.mosquitto.org', '');

Future<int> mqqtClient() async {

  client.logging(on: false);

非默认消息（60s） /// 你必须在这里设置它 client.keepAlivePeriod = 20;

  /// Add the unsolicited disconnection callback
  client.onDisconnected = onDisconnected;

  /// Add the successful connection callback
  client.onConnected = onConnected;

  client.onSubscribed = onSubscribed;


  client.pongCallback = pong;

  /// Create a connection message to use or use the default one. The default one sets the
  /// client identifier, any supplied username/password, the default keepalive interval(60s)
  /// and clean session, an example of a specific one below.
  final MqttConnectMessage connMess = MqttConnectMessage()
      .withClientIdentifier('Mqtt_MyClientUniqueId')
      .keepAliveFor(20) // Must agree with the keep alive set above or not set
      .withWillTopic('willtopic') // If you set this you must set a will message
      .withWillMessage('My Will message')
      .startClean() // Non persistent session for testing
      .withWillQos(MqttQos.atLeastOnce);
  print('EXAMPLE::Mosquitto client connecting....');
  client.connectionMessage = connMess;


  try {
    await client.connect();
  } on Exception catch (e) {
    print('EXAMPLE::client exception - $e');
    client.disconnect();
  }

  /// Check we are connected
  if (client.connectionStatus.state == MqttConnectionState.connected) {
    print('EXAMPLE::Mosquitto client connected');
  } else {
    /// Use status here rather than state if you also want the broker return code.
    print(
        'EXAMPLE::ERROR Mosquitto client connection failed - disconnecting, status is ${client.connectionStatus}');
    client.disconnect();
    exit(-1);
  }

  /// Ok, lets try a subscription
  print('EXAMPLE::Subscribing to the test/lol topic');
  const String topic = 'test/lol'; // Not a wildcard topic
  client.subscribe(topic, MqttQos.atMostOnce);

  /// The client has a change notifier object(see the Observable class) which we then listen to to get
  /// notifications of published updates to each subscribed topic.
  client.updates.listen((List<MqttReceivedMessage<MqttMessage>> c) {
    final MqttPublishMessage recMess = c[0].payload;
    final String pt =
    MqttPublishPayload.bytesToStringAsString(recMess.payload.message);


    print(
        'EXAMPLE::Change notification:: topic is <${c[0].topic}>, payload is <-- $pt -->');
    print('');
  });

  /// If needed you can listen for published messages that have completed the publishing
  /// handshake which is Qos dependant. Any message received on this stream has completed its
  /// publishing handshake with the broker.
  client.published.listen((MqttPublishMessage message) {
    print(
        'EXAMPLE::Published notification:: topic is ${message.variableHeader.topicName}, with Qos ${message.header.qos}');
  });

  /// Lets publish to our topic
  /// Use the payload builder rather than a raw buffer
  /// Our known topic to publish to
  const String pubTopic = 'Dart/Mqtt_client/testtopic';
  final MqttClientPayloadBuilder builder = MqttClientPayloadBuilder();
  builder.addString('Hello from mqtt_client');

  /// Subscribe to it
  print('EXAMPLE::Subscribing to the Dart/Mqtt_client/testtopic topic');
  client.subscribe(pubTopic, MqttQos.exactlyOnce);

  /// Publish it
  print('EXAMPLE::Publishing our topic');
  client.publishMessage(pubTopic, MqttQos.exactlyOnce, builder.payload);

  /// Ok, we will now sleep a while, in this gap you will see ping request/response
  /// messages being exchanged by the keep alive mechanism.
  print('EXAMPLE::Sleeping....');
  await MqttUtilities.asyncSleep(120);

  /// Finally, unsubscribe and exit gracefully
  print('EXAMPLE::Unsubscribing');
  client.unsubscribe(topic);

  /// Wait for the unsubscribe message from the broker if you wish.
  await MqttUtilities.asyncSleep(2);
  print('EXAMPLE::Disconnecting');
  client.disconnect();
  return 0;
}

/// The subscribed callback
void onSubscribed(String topic) {
  print('EXAMPLE::Subscription confirmed for topic $topic');
}

/// The unsolicited disconnect callback
void onDisconnected() {
  print('EXAMPLE::OnDisconnected client callback - Client disconnection');
  if (client.connectionStatus.returnCode == MqttConnectReturnCode.solicited) {
    print('EXAMPLE::OnDisconnected callback is solicited, this is correct');
  }
  exit(-1);
}

/// The successful connect callback
void onConnected() {
  print(
      'EXAMPLE::OnConnected client callback - Client connection was sucessful');
}

/// Pong callback
void pong() {
  print('EXAMPLE::Ping response client callback invoked');
}

标签： flutterdartmqtt

解决方案

您的 Flutter 版本使用的是 Dart 2.5，MQTT 客户端仍在 Dart 2.4.x 上，因为这是当前的稳定版本，您可以使用 mqqt_client repo 的开发分支，请参阅此问题

推荐阅读