uniapp-mqtt-onenet/uni_modules/flight-mqtt/utssdk/app-android/index.uts

import MqttClient from 'org.eclipse.paho.client.mqttv3.MqttClient';
import MqttConnectOptions from 'org.eclipse.paho.client.mqttv3.MqttConnectOptions';
import MqttMessage from 'org.eclipse.paho.client.mqttv3.MqttMessage';
import MemoryPersistence from 'org.eclipse.paho.client.mqttv3.persist.MemoryPersistence';
import MqttCallbackExtended from 'org.eclipse.paho.client.mqttv3.MqttCallbackExtended';
import IMqttDeliveryToken from 'org.eclipse.paho.client.mqttv3.IMqttDeliveryToken';

class MyMqttCallback implements MqttCallbackExtended {
  private manager:MyMqttManager
   
   constructor(t:MyMqttManager){
     this.manager = t
   }

    override connectionLost(cause : Throwable | null) {
        console.error("MQTT Lost: " + (cause != null ? cause.message : "unknown"));
    }

    override messageArrived(topic : string, message : MqttMessage) {
        // 1. 获取字节数组
        const payload = message.getPayload();
        
        // 2. 转换为 UTF-8 字符串 (这是最稳妥的 Java 原生做法)
        const content = new java.lang.String(payload, "UTF-8").toString();
        
        console.log("收到主题:", topic);
        console.log("收到内容:", content, this.manager.onMessageReceived);
    
        // 3. 回传给 Vue (注意：此处在非 UI 线程，Vue 内部会自动处理)
        if (this.manager.onMessageReceived != null) {
            this.manager.onMessageReceived!(topic, content);
        }
    }

    override deliveryComplete(token : IMqttDeliveryToken | null) {
      console.log(token)
        // 可留空
    }
    
    override connectComplete(reconnect:boolean, serverURI:string)
    {
      console.log("链接完成:", serverURI, reconnect)
      this.manager.reconnect()
    }
}

export class MyMqttManager {
    private client: MqttClient | null = null;
    private brokerUrl: string = ""
    private clientId: string = ""
    private username: string = ""
    private password: string = ""
    
    private topics: string[] = []
    
    // 定义一个回调函数，用于把消息传回 Vue 界面
    public onMessageReceived : ((topic : string, message : string) => void) | null = null;
    
    on(on:((topic : string, message : string) => void) | null){
      this.onMessageReceived = on
    }
    
    reconnect(){
      // this.connect(this.brokerUrl, this.clientId, this.username, this.password)
      this.topics.forEach((t : string) => {
          this.client!.subscribe(t);
          console.log("正在订阅单个主题:", t);
      });
    }

    initCallback() {
        if (this.client == null) return;

        // 1. 先声明实例
        const callbackInstance = new MyMqttCallback(this); // <--- 注意这里的括号，表示实例化
        
        // 2. 再进行设置
        this.client!.setCallback(callbackInstance);
    }
    
    /**
     * 连接 MQTT 服务器
     */
    connect(broker: string, clientId: string, username:string, password: string) {
        // #ifdef APP-ANDROID
        try {
            this.clientId = clientId
            this.username = username
            this.password = password
            const persistence = new MemoryPersistence();
            this.client = new MqttClient(broker, this.clientId, persistence);
            
            const connOpts = new MqttConnectOptions();
            connOpts.setCleanSession(true);
            connOpts.setConnectionTimeout(10);
            connOpts.setUserName(this.username)
            connOpts.setPassword(this.password.toCharArray())
            // connOpts.setKeepAliveInterval(60)
            
            console.log("正在连接 MQTT: " + broker);
            connOpts.setAutomaticReconnect(true)
            this.client!.connect(connOpts);
            console.log("MQTT 连接成功");
            
        } catch (e) {
            console.error("MQTT 连接失败: " + e);
        }
        // #endif
    }

    /**
     * 发布消息
     */
    publish(topic: string, content: string) {
        // #ifdef APP-ANDROID
        try {
            if (this.client != null && this.client!.isConnected()) {
                const message = new MqttMessage(content.toByteArray());
                message.setQos(1);
                this.client!.publish(topic, message);
                console.log("消息发送成功", topic, content);
            }
        } catch (e) {
            console.error("发布失败: " + e);
            throw e
        }
        // #endif
    }
    
    /**
     * 订阅消息
     */
    subscribe(topic: string) {
        try {
            this.topics.push(topic)
            if (this.client != null && this.client!.isConnected()) {
                // Paho 的 subscribe 需要 Int 类型
                this.client!.subscribe(topic);
                console.log("订阅成功: " + topic);
            } else {
                console.error("订阅失败: 客户端未连接");
            }
        } catch (e) {
            console.error("订阅操作异常: " + e);
        }
    }
    
    /**
     * 关闭 mqtt
     */
    close(){
      try{
        if (this.client != null && this.client!.isConnected()) {
            // Paho 的 subscribe 需要 Int 类型
            this.client!.close();
            console.log("关闭成功: ");
        } else {
            console.error("关闭失败: 客户端未连接");
        }
      }catch(e){
        console.error("关闭操作异常: " + e);
      }
    }
}