diff --git a/app/build.gradle b/app/build.gradle index f530f37..88146c2 100644 --- a/app/build.gradle +++ b/app/build.gradle @@ -93,6 +93,11 @@ android { aidl true } + packagingOptions { + exclude 'META-INF/io.netty.versions.properties' + exclude 'META-INF/INDEX.LIST' + } + externalNativeBuild { cmake { path file('CMakeLists.txt') @@ -222,6 +227,9 @@ dependencies { implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-android:1.11.0' implementation 'org.jetbrains.kotlinx:kotlinx-coroutines-rx3:1.11.0' + // HiveMQ MQTT 客户端 (MQTT 3.1.1 / 5.0) + implementation 'com.hivemq:hivemq-mqtt-client:1.3.17' + //保持1.3.1 更新会报错 implementation 'androidx.appcompat:appcompat:1.7.1' //2.0.4以上无法预览 diff --git a/app/proguard-rules.pro b/app/proguard-rules.pro index f1b4245..9252dec 100644 --- a/app/proguard-rules.pro +++ b/app/proguard-rules.pro @@ -19,3 +19,7 @@ # If you keep the line number information, uncomment this to # hide the original source file name. #-renamesourcefileattribute SourceFile + +# ===== HiveMQ MQTT Client (依赖 Netty、JCTools) ===== +-keepclassmembernames class io.netty.** { *; } +-keepclassmembers class org.jctools.** { *; } diff --git a/app/src/main/java/com/ttstd/dialer/base/BaseApplication.java b/app/src/main/java/com/ttstd/dialer/base/BaseApplication.java index 3aebfbf..c620bab 100644 --- a/app/src/main/java/com/ttstd/dialer/base/BaseApplication.java +++ b/app/src/main/java/com/ttstd/dialer/base/BaseApplication.java @@ -23,12 +23,14 @@ import com.ttstd.dialer.config.SystemIntentAction; import com.ttstd.dialer.data.cache.CacheManager; import com.ttstd.dialer.manager.AppManager; import com.ttstd.dialer.manager.MapManager; +import com.ttstd.dialer.manager.MqttManager; import com.ttstd.dialer.manager.WeatherManager; import com.ttstd.dialer.mdm.DeviceManagerService; import com.ttstd.dialer.network.OkHttpManager; import com.ttstd.dialer.push.PushExecutor; import com.ttstd.dialer.receiver.AppChangedReceiver; import com.ttstd.dialer.receiver.HourlyChimeManager; +import com.ttstd.dialer.receiver.ScreenStateReceiver; import com.ttstd.dialer.tts.sherpa_onnx.SherpaOnnxTtsManager; import com.ttstd.dialer.utils.Logger; import com.ttstd.dialer.utils.NativeUtils; @@ -130,6 +132,11 @@ public class BaseApplication extends Application { WeatherManager.init(this); IconCacheManager.init(this); + + // 初始化 MQTT 并连接 Broker + MqttManager.init(this); + MqttManager.getInstance().connect(); + MqttManager.getInstance().subscribe(CommonConfig.MQTT_TOPIC_COMMAND, 1); SherpaOnnxTtsManager.getInstance().init(this); AlarmManagerHelper.rescheduleAllAlarms(this); @@ -138,6 +145,8 @@ public class BaseApplication extends Application { } registerReceivers(); + // 注册屏幕亮灭监听，上报设备亮屏/熄屏状态 + ScreenStateReceiver.register(this); // 启动时异步清理磁盘缓存（过期淘汰 + 容量上限），不阻塞启动 CacheManager.getInstance(this).cleanupAsync(); diff --git a/app/src/main/java/com/ttstd/dialer/config/CommonConfig.java b/app/src/main/java/com/ttstd/dialer/config/CommonConfig.java index f86225b..671c3f6 100644 --- a/app/src/main/java/com/ttstd/dialer/config/CommonConfig.java +++ b/app/src/main/java/com/ttstd/dialer/config/CommonConfig.java @@ -3,6 +3,16 @@ package com.ttstd.dialer.config; public class CommonConfig { public static final String MMKV_ID = "InterProcessKV"; + /*MQTT 配置*/ + public static final String MQTT_URL = "mqtt://175.178.213.60:1883"; + public static final String MQTT_CLIENT_ID = "ttstd-dialer-android"; + public static final String MQTT_USERNAME = ""; + public static final String MQTT_PASSWORD = ""; + /*订阅主题*/ + public static final String MQTT_TOPIC_COMMAND = "ttstd/device/command"; + /*发布主题*/ + public static final String MQTT_TOPIC_STATUS = "ttstd/device/status"; + public static final String CONTACT_HOME_PAGE = "contact_home_page_key"; public static final String FLOAT_WINDOW_X = "float_window_x_key"; diff --git a/app/src/main/java/com/ttstd/dialer/manager/MqttManager.java b/app/src/main/java/com/ttstd/dialer/manager/MqttManager.java new file mode 100644 index 0000000..9161de1 --- /dev/null +++ b/app/src/main/java/com/ttstd/dialer/manager/MqttManager.java @@ -0,0 +1,481 @@ +package com.ttstd.dialer.manager; + +import android.annotation.SuppressLint; +import android.content.Context; +import android.os.Handler; +import android.os.Looper; +import android.text.TextUtils; + +import com.hivemq.client.mqtt.mqtt5.Mqtt5AsyncClient; +import com.hivemq.client.mqtt.mqtt5.Mqtt5Client; +import com.hivemq.client.mqtt.mqtt5.Mqtt5ClientBuilder; +import com.jeremyliao.liveeventbus.LiveEventBus; +import com.ttstd.dialer.config.CommonConfig; +import com.ttstd.dialer.mdm.DeviceManagerService; +import com.ttstd.dialer.utils.Logger; +import com.ttstd.dialer.utils.SystemUtils; + +import java.net.Inet4Address; +import java.net.InetAddress; +import java.net.NetworkInterface; +import java.net.URI; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.Enumeration; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; + +/** + * MQTT 消息管理 + *

+ * 基于 HiveMQ MQTT Client (MQTT 5.0/3.1.1) 封装与 Broker 的连接、订阅与发布。 + * 连接成功后自动重连，收到订阅主题的消息通过 LiveEventBus 广播。 + * + * @author TTSTD + * @since 2026/8/8 + */ +public class MqttManager { + private static final String TAG = "MqttManager"; + + /** + * 收到 MQTT 消息的事件 key，消息体为 {@link MqttMessage} + */ + public static final String EVENT_MQTT_MESSAGE = "mqtt_message"; + + /** + * 连接状态变化事件 key，消息体为 Boolean (true=已连接, false=断开) + */ + public static final String EVENT_MQTT_CONNECT_STATE = "mqtt_connect_state"; + + /** + * 心跳间隔(毫秒)，后端据此判定设备在线/离线(后端 online-timeout 应大于此值) + */ + private static final long HEARTBEAT_INTERVAL_MS = 30_000L; + + @SuppressLint("StaticFieldLeak") + private static volatile MqttManager INSTANCE; + + private final Context mContext; + private Mqtt5AsyncClient mMqttClient; + private volatile boolean mConnected; + + /** + * 主动断开标记，避免主动断开时触发自动重连 + */ + private volatile boolean mManualDisconnect; + + /** + * 心跳定时器（主线程 Handler） + */ + private final Handler mHeartbeatHandler = new Handler(Looper.getMainLooper()); + + /** + * 重连定时器（主线程 Handler） + */ + private final Handler mReconnectHandler = new Handler(Looper.getMainLooper()); + + /** + * 重连退避间隔（毫秒），随失败次数指数增长 + */ + private long mReconnectDelayMs = 1_000L; + + private boolean mHeartbeatRunning; + + /** + * 已订阅的主题 + */ + private final Map mSubscribedTopics = new ConcurrentHashMap<>(); + + private final Runnable mHeartbeatTask = new Runnable() { + @Override + public void run() { + if (!mConnected) { + return; + } + reportHeartbeat(); + mHeartbeatHandler.postDelayed(this, HEARTBEAT_INTERVAL_MS); + } + }; + + private MqttManager(Context context) { + this.mContext = context.getApplicationContext(); + } + + public static void init(Context context) { + if (INSTANCE == null) { + synchronized (MqttManager.class) { + if (INSTANCE == null) { + INSTANCE = new MqttManager(context); + } + } + } + } + + public static MqttManager getInstance() { + if (INSTANCE == null) { + throw new IllegalStateException("You must first initialize the MqttManager"); + } + return INSTANCE; + } + + /** + * 建立 MQTT 连接（异步，断线后自动重连） + */ + public void connect() { + try { + if (mMqttClient != null && mConnected) { + return; + } + // 主动发起连接，清除主动断开标记，允许后续自动重连 + mManualDisconnect = false; + if (mMqttClient != null) { + try { + mMqttClient.disconnect(); + } catch (Throwable ignore) { + } + mMqttClient = null; + } + + URI uri = URI.create(CommonConfig.MQTT_URL); + String host = uri.getHost(); + int port = uri.getPort() > 0 ? uri.getPort() : 1883; + + Mqtt5ClientBuilder builder = Mqtt5Client.builder() + .identifier(CommonConfig.MQTT_CLIENT_ID) + .serverHost(host) + .serverPort(port) + .addDisconnectedListener(context -> onDisconnected()); + if (!TextUtils.isEmpty(CommonConfig.MQTT_USERNAME)) { + builder.simpleAuth() + .username(CommonConfig.MQTT_USERNAME) + .password(CommonConfig.MQTT_PASSWORD.getBytes(StandardCharsets.UTF_8)) + .applySimpleAuth(); + } + + mMqttClient = builder.buildAsync(); + + mMqttClient.connectWith() + .keepAlive(30) + .send() + .whenComplete((connAck, throwable) -> { + if (throwable != null) { + String msg = throwable.getMessage(); + Logger.e(TAG, "connect: 连接失败 " + msg + + "，将自动重试连接 " + host + ":" + port); + setConnected(false); + // 首次连接失败也安排重连（网络不可达等场景） + scheduleReconnect(); + } else { + Logger.e(TAG, "connect: 连接成功"); + mReconnectDelayMs = 1_000L; + setConnected(true); + // 连接成功后重新订阅之前订阅的主题 + reSubscribeAll(); + // 连接成功后启动心跳上报 + startHeartbeat(); + // 连接成功后上报一次当前屏幕开关状态 + reportScreenStateOnConnect(); + } + }); + } catch (Throwable e) { + Logger.e(TAG, "connect: 异常 " + e.getMessage()); + e.printStackTrace(); + } + } + + /** + * 断线回调：若非主动断开，则按指数退避安排重连 + */ + private void onDisconnected() { + if (mManualDisconnect) { + Logger.e(TAG, "onDisconnected: 主动断开，不重连"); + return; + } + Logger.e(TAG, "onDisconnected: 连接断开，准备自动重连"); + setConnected(false); + stopHeartbeat(); + scheduleReconnect(); + } + + /** + * 安排一次延迟重连（指数退避，上限 30s） + */ + private void scheduleReconnect() { + mReconnectHandler.removeCallbacks(mReconnectRunnable); + Logger.e(TAG, "scheduleReconnect: " + (mReconnectDelayMs / 1000) + "s 后重连"); + mReconnectHandler.postDelayed(mReconnectRunnable, mReconnectDelayMs); + // 下次退避间隔翻倍，封顶 30s + mReconnectDelayMs = Math.min(mReconnectDelayMs * 2, 30_000L); + } + + private final Runnable mReconnectRunnable = new Runnable() { + @Override + public void run() { + if (mManualDisconnect) { + return; + } + Logger.e(TAG, "reconnect: 尝试重新连接..."); + connect(); + } + }; + + /** + * 断开 MQTT 连接（主动断开，不会触发自动重连） + */ + public void disconnect() { + // 标记主动断开，重连任务与断线监听均据此跳过重连 + mManualDisconnect = true; + mReconnectHandler.removeCallbacks(mReconnectRunnable); + stopHeartbeat(); + try { + if (mMqttClient != null) { + mMqttClient.disconnect(); + mMqttClient = null; + } + setConnected(false); + } catch (Throwable e) { + Logger.e(TAG, "disconnect: " + e.getMessage()); + } + } + + /** + * 是否已连接 + */ + public boolean isConnected() { + return mConnected; + } + + /** + * 订阅主题，收到消息通过 {@link #EVENT_MQTT_MESSAGE} 广播 + * + * @param topic 主题 + * @param qos QoS (0-2) + */ + public void subscribe(String topic, int qos) { + if (TextUtils.isEmpty(topic) || mMqttClient == null) { + Logger.e(TAG, "subscribe: 客户端未就绪，跳过订阅 " + topic); + return; + } + mSubscribedTopics.put(topic, topic); + mMqttClient.subscribeWith() + .topicFilter(topic) + .qos(convertQos(qos)) + .callback(this::onMessageReceived) + .send() + .whenComplete((subAck, throwable) -> { + if (throwable != null) { + Logger.e(TAG, "subscribe: 订阅失败 " + topic + " " + throwable.getMessage()); + } else { + Logger.e(TAG, "subscribe: 订阅成功 " + topic); + } + }); + } + + /** + * 取消订阅 + */ + public void unsubscribe(String topic) { + mSubscribedTopics.remove(topic); + if (mMqttClient == null) { + return; + } + try { + mMqttClient.unsubscribeWith().topicFilter(topic).send(); + } catch (Throwable e) { + Logger.e(TAG, "unsubscribe: " + e.getMessage()); + } + } + + /** + * 发布消息 + * + * @param topic 主题 + * @param payload 消息内容 + */ + public void publish(String topic, String payload) { + publish(topic, payload.getBytes(StandardCharsets.UTF_8)); + } + + /** + * 发布消息 + * + * @param topic 主题 + * @param payload 消息内容(字节) + */ + public void publish(String topic, byte[] payload) { + if (mMqttClient == null || !mConnected) { + Logger.e(TAG, "publish: 客户端未连接，丢弃消息 topic=" + topic); + return; + } + mMqttClient.publishWith() + .topic(topic) + .payload(payload) + .send() + .whenComplete((publishResult, throwable) -> { + if (throwable != null) { + Logger.e(TAG, "publish: 发布失败 topic=" + topic + " " + throwable.getMessage()); + } else { + Logger.e(TAG, "publish: 发布成功 topic=" + topic); + } + }); + } + + /** + * 上报设备状态，默认发布到 {@link CommonConfig#MQTT_TOPIC_STATUS} + */ + public void publishStatus(String payload) { + publish(CommonConfig.MQTT_TOPIC_STATUS, payload); + } + + /** + * 启动心跳上报（连接成功后调用）。后端收到心跳刷新 Redis 在线状态 TTL。 + */ + public void startHeartbeat() { + if (mHeartbeatRunning) { + return; + } + mHeartbeatRunning = true; + // 立即上报一次，然后周期性上报 + mHeartbeatTask.run(); + } + + /** + * 停止心跳上报 + */ + public void stopHeartbeat() { + mHeartbeatRunning = false; + mHeartbeatHandler.removeCallbacks(mHeartbeatTask); + } + + /** + * 立即上报一次心跳，消息体为 JSON：{"sn":"xxx","status":"online","ip":"x","screenOn":true|false} + */ + public void reportHeartbeat() { + String sn = DeviceManagerService.getInstance().getSerial(); + if (TextUtils.isEmpty(sn)) { + Logger.e(TAG, "reportHeartbeat: 设备序列号为空，跳过心跳"); + return; + } + publish(CommonConfig.MQTT_TOPIC_STATUS, buildStatusPayload(sn, SystemUtils.isScreenOn(mContext))); + } + + /** + * 上报设备屏幕亮灭状态，消息体为 JSON： + * {"sn":"xxx","status":"online","ip":"x","screenOn":true|false} + * + * @param screenOn true=亮屏, false=熄屏 + */ + public void reportScreenState(boolean screenOn) { + String sn = DeviceManagerService.getInstance().getSerial(); + if (TextUtils.isEmpty(sn)) { + Logger.e(TAG, "reportScreenState: 设备序列号为空，跳过上报"); + return; + } + publish(CommonConfig.MQTT_TOPIC_STATUS, buildStatusPayload(sn, screenOn)); + } + + /** + * 连接成功时上报一次当前屏幕真实开关状态。 + */ + public void reportScreenStateOnConnect() { + boolean screenOn = SystemUtils.isScreenOn(mContext); + Logger.e(TAG, "reportScreenStateOnConnect: screenOn=" + screenOn); + reportScreenState(screenOn); + } + + /** + * 构建设备状态上报 JSON：{"sn":"...","status":"online","ip":"...","screenOn":true|false} + */ + private String buildStatusPayload(String sn, boolean screenOn) { + String ip = getLocalIpAddress(); + return "{\"sn\":\"" + sn + "\",\"status\":\"online\",\"ip\":\"" + (ip == null ? "" : ip) + + "\",\"screenOn\":" + screenOn + "}"; + } + + /** + * 获取本地 IP 地址（获取失败返回空串） + */ + private String getLocalIpAddress() { + try { + Enumeration interfaces = NetworkInterface.getNetworkInterfaces(); + while (interfaces.hasMoreElements()) { + NetworkInterface networkInterface = interfaces.nextElement(); + for (InetAddress inetAddress : Collections.list(networkInterface.getInetAddresses())) { + if (!inetAddress.isLoopbackAddress() && inetAddress instanceof Inet4Address) { + return inetAddress.getHostAddress(); + } + } + } + } catch (Throwable e) { + Logger.e(TAG, "getLocalIpAddress: " + e.getMessage()); + } + return ""; + } + + private void onMessageReceived(com.hivemq.client.mqtt.mqtt5.message.publish.Mqtt5Publish publish) { + String topic = publish.getTopic().toString(); + String payload = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8); + Logger.e(TAG, "onMessageReceived: topic=" + topic + ", payload=" + payload); + LiveEventBus.get(EVENT_MQTT_MESSAGE) + .post(new MqttMessage(topic, payload)); + } + + private void reSubscribeAll() { + if (mSubscribedTopics.isEmpty()) { + return; + } + Logger.e(TAG, "reSubscribeAll: 重新订阅 " + mSubscribedTopics.size() + " 个主题"); + for (String topic : mSubscribedTopics.keySet()) { + mMqttClient.subscribeWith() + .topicFilter(topic) + .qos(convertQos(1)) + .callback(this::onMessageReceived) + .send(); + } + } + + private void setConnected(boolean connected) { + if (mConnected != connected) { + mConnected = connected; + LiveEventBus.get(EVENT_MQTT_CONNECT_STATE).post(connected); + } + } + + private com.hivemq.client.mqtt.datatypes.MqttQos convertQos(int qos) { + switch (qos) { + case 2: + return com.hivemq.client.mqtt.datatypes.MqttQos.EXACTLY_ONCE; + case 0: + return com.hivemq.client.mqtt.datatypes.MqttQos.AT_MOST_ONCE; + case 1: + default: + return com.hivemq.client.mqtt.datatypes.MqttQos.AT_LEAST_ONCE; + } + } + + /** + * MQTT 消息封装 + */ + public static class MqttMessage { + private final String topic; + private final String payload; + + public MqttMessage(String topic, String payload) { + this.topic = topic; + this.payload = payload; + } + + public String getTopic() { + return topic; + } + + public String getPayload() { + return payload; + } + + @Override + public String toString() { + return "MqttMessage{topic='" + topic + "', payload='" + payload + "'}"; + } + } +} diff --git a/app/src/main/java/com/ttstd/dialer/receiver/ScreenStateReceiver.java b/app/src/main/java/com/ttstd/dialer/receiver/ScreenStateReceiver.java new file mode 100644 index 0000000..9976415 --- /dev/null +++ b/app/src/main/java/com/ttstd/dialer/receiver/ScreenStateReceiver.java @@ -0,0 +1,54 @@ +package com.ttstd.dialer.receiver; + +import android.content.BroadcastReceiver; +import android.content.Context; +import android.content.Intent; +import android.content.IntentFilter; + +import com.ttstd.dialer.manager.MqttManager; +import com.ttstd.dialer.utils.Logger; + +/** + * 屏幕亮灭状态监听 + *

+ * 监听 {@link Intent#ACTION_SCREEN_ON} 与 {@link Intent#ACTION_SCREEN_OFF}， + * 通过 MQTT 向服务端上报设备亮屏/熄屏状态（{@code screenOn} 字段）。 + * + * @author TTSTD + * @since 2026/8/8 + */ +public class ScreenStateReceiver extends BroadcastReceiver { + + private static final String TAG = "ScreenStateReceiver"; + + /** 注册屏幕亮灭广播 */ + public static void register(Context context) { + IntentFilter filter = new IntentFilter(); + filter.addAction(Intent.ACTION_SCREEN_ON); + filter.addAction(Intent.ACTION_SCREEN_OFF); + context.registerReceiver(new ScreenStateReceiver(), filter); + // 注册时上报一次当前状态 + report(true); + } + + @Override + public void onReceive(Context context, Intent intent) { + if (intent == null || intent.getAction() == null) { + return; + } + boolean screenOn = Intent.ACTION_SCREEN_ON.equals(intent.getAction()); + Logger.e(TAG, "onReceive: action=" + intent.getAction() + ", screenOn=" + screenOn); + report(screenOn); + } + + /** + * 上报屏幕状态（1=亮屏, 0=熄屏） + */ + private static void report(boolean screenOn) { + try { + MqttManager.getInstance().reportScreenState(screenOn); + } catch (Throwable e) { + Logger.e(TAG, "report: 上报屏幕状态失败 " + e.getMessage()); + } + } +} diff --git a/app/src/main/java/com/ttstd/dialer/utils/SystemUtils.java b/app/src/main/java/com/ttstd/dialer/utils/SystemUtils.java index 0114c2d..3bcccc4 100644 --- a/app/src/main/java/com/ttstd/dialer/utils/SystemUtils.java +++ b/app/src/main/java/com/ttstd/dialer/utils/SystemUtils.java @@ -23,6 +23,7 @@ import android.net.wifi.WifiManager; import android.os.Build; import android.os.Environment; import android.os.IBinder; +import android.os.PowerManager; import android.os.Process; import android.os.RemoteException; import android.os.UserHandle; @@ -1078,7 +1079,7 @@ public class SystemUtils { // 这是一个针对 Android 9/10 的简化逻辑参考： Bitmap bitmap = (Bitmap) surfaceControlClass.getDeclaredMethod("screenshot", - Rect.class, Integer.TYPE, Integer.TYPE, Integer.TYPE) + Rect.class, Integer.TYPE, Integer.TYPE, Integer.TYPE) .invoke(null, new Rect(), screenWidth, screenHeight, 0); return bitmap; @@ -1244,4 +1245,16 @@ public class SystemUtils { } } + /** + * 查询设备当前屏幕是否亮屏。 + */ + public static boolean isScreenOn(Context context) { + try { + PowerManager powerManager = + (PowerManager) context.getSystemService(Context.POWER_SERVICE); + return powerManager != null && powerManager.isInteractive(); + } catch (Throwable e) { + return false; + } + } }