tongtong/ElderlyDialer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(base): 集成 MQTT 并添加屏幕状态监听

Browse Source 
- 初始化 HiveMQ MQTT 客户端并连接 Broker，订阅命令主题
- 添加屏幕亮灭广播监听，支持上报设备亮屏/熄屏状态
- 新增 MQTT 配置（地址、主题等）及 SystemUtils 屏幕状态查询方法
- 添加 HiveMQ 依赖并配置 ProGuard 混淆规则与打包排除项
This commit is contained in:
TongTongStudio
2026-08-08 20:36:00 +08:00
parent f2e22c1fb6
commit ce34353ae3
7 changed files with 580 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
app/build.gradle
View File

@@ -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以上无法预览

4
app/proguard-rules.pro vendored
View File

@@ -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.** { *; }

9
app/src/main/java/com/ttstd/dialer/base/BaseApplication.java
View File

@@ -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();

10
app/src/main/java/com/ttstd/dialer/config/CommonConfig.java
View File

@@ -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";

481
app/src/main/java/com/ttstd/dialer/manager/MqttManager.java Normal file
View File

@@ -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 消息管理
* <p>
* 基于 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<String, String> 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<NetworkInterface> 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 + "'}";
}
}
}

54
app/src/main/java/com/ttstd/dialer/receiver/ScreenStateReceiver.java Normal file
View File

@@ -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;
/**
* 屏幕亮灭状态监听
* <p>
* 监听 {@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());
}
}
}

15
app/src/main/java/com/ttstd/dialer/utils/SystemUtils.java
View File

@@ -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;
}
}
}