jiahong
443 lines
13 KiB
Kotlin
443 lines
13 KiB
Kotlin
/**
|
||
* 自然写互动课堂智慧黑板端应用软件 V1.0
|
||
*
|
||
* StrokeReceiver.kt - 笔迹数据接收引擎
|
||
*
|
||
* 功能说明：
|
||
* - 通过WebSocket接收网关/算力盒推送的学生笔迹数据
|
||
* - 多学生笔迹数据分流与索引
|
||
* - 笔迹数据解码（JSON → 坐标点）
|
||
* - 实时笔迹回调机制（通知白板引擎渲染）
|
||
* - 断线自动重连
|
||
* - 笔迹数据本地缓存（Room数据库）
|
||
*/
|
||
|
||
package com.writech.board.engine
|
||
|
||
import android.util.Log
|
||
import org.json.JSONArray
|
||
import org.json.JSONObject
|
||
import java.net.URI
|
||
import java.util.concurrent.*
|
||
import java.util.concurrent.atomic.AtomicBoolean
|
||
import java.util.concurrent.atomic.AtomicLong
|
||
|
||
/**
|
||
* 学生笔迹数据包
|
||
*/
|
||
data class StudentStrokeData(
|
||
val studentId: String, /* 学生ID */
|
||
val penId: String, /* 笔MAC地址 */
|
||
val points: List<StrokePoint>, /* 坐标点列表 */
|
||
val pageId: Int = 0, /* 页面ID */
|
||
val isPenDown: Boolean = true, /* 落笔/抬笔状态 */
|
||
val timestamp: Long = System.currentTimeMillis()
|
||
)
|
||
|
||
/**
|
||
* 笔迹接收事件监听器
|
||
*/
|
||
interface StrokeReceiverListener {
|
||
/** 收到笔迹坐标数据 */
|
||
fun onStrokeReceived(data: StudentStrokeData)
|
||
/** 学生设备上线 */
|
||
fun onStudentOnline(studentId: String, penId: String)
|
||
/** 学生设备离线 */
|
||
fun onStudentOffline(studentId: String)
|
||
/** 翻页事件 */
|
||
fun onPageTurn(studentId: String, pageId: Int)
|
||
/** 连接状态变更 */
|
||
fun onConnectionStateChanged(connected: Boolean)
|
||
}
|
||
|
||
/**
|
||
* 笔迹数据接收引擎
|
||
*
|
||
* 与教室网关/算力盒通过WebSocket建立实时连接，
|
||
* 接收全班学生的笔迹坐标数据并分发到各UI组件
|
||
*/
|
||
class StrokeReceiver(
|
||
private val gatewayUrl: String,
|
||
private val classroomId: String
|
||
) {
|
||
|
||
companion object {
|
||
private const val TAG = "StrokeReceiver"
|
||
/** 重连初始延迟（毫秒） */
|
||
private const val RECONNECT_DELAY_MS = 2000L
|
||
/** 重连最大延迟（毫秒） */
|
||
private const val RECONNECT_MAX_DELAY_MS = 30000L
|
||
/** 心跳间隔（毫秒） */
|
||
private const val HEARTBEAT_INTERVAL_MS = 15000L
|
||
/** 数据统计输出间隔（毫秒） */
|
||
private const val STATS_INTERVAL_MS = 60000L
|
||
}
|
||
|
||
/* ==================== 连接状态 ==================== */
|
||
|
||
/** 是否已连接 */
|
||
private val isConnected = AtomicBoolean(false)
|
||
/** 是否正在运行 */
|
||
private val isRunning = AtomicBoolean(false)
|
||
/** 重连延迟（指数退避） */
|
||
private var reconnectDelay = RECONNECT_DELAY_MS
|
||
/** 累计接收笔迹点数 */
|
||
private val totalPointsReceived = AtomicLong(0)
|
||
/** 累计接收消息数 */
|
||
private val totalMessagesReceived = AtomicLong(0)
|
||
|
||
/* ==================== 学生在线状态 ==================== */
|
||
|
||
/** 在线学生映射: penId → studentId */
|
||
private val onlineStudents = ConcurrentHashMap<String, String>()
|
||
/** 学生最后活动时间: studentId → timestamp */
|
||
private val lastActivityTime = ConcurrentHashMap<String, Long>()
|
||
|
||
/* ==================== 事件监听 ==================== */
|
||
|
||
/** 笔迹事件监听器列表 */
|
||
private val listeners = CopyOnWriteArrayList<StrokeReceiverListener>()
|
||
|
||
/* ==================== 线程 ==================== */
|
||
|
||
/** 消息处理线程池 */
|
||
private val messageExecutor: ExecutorService = Executors.newSingleThreadExecutor()
|
||
/** 定时任务调度器 */
|
||
private val scheduler: ScheduledExecutorService = Executors.newScheduledThreadPool(1)
|
||
|
||
/**
|
||
* 添加事件监听器
|
||
*/
|
||
fun addListener(listener: StrokeReceiverListener) {
|
||
listeners.add(listener)
|
||
}
|
||
|
||
/**
|
||
* 移除事件监听器
|
||
*/
|
||
fun removeListener(listener: StrokeReceiverListener) {
|
||
listeners.remove(listener)
|
||
}
|
||
|
||
/**
|
||
* 启动笔迹接收
|
||
* 连接WebSocket并开始接收数据
|
||
*/
|
||
fun start() {
|
||
if (isRunning.getAndSet(true)) {
|
||
Log.w(TAG, "接收器已在运行")
|
||
return
|
||
}
|
||
|
||
Log.i(TAG, "启动笔迹接收, 网关=$gatewayUrl, 教室=$classroomId")
|
||
|
||
/* 建立WebSocket连接 */
|
||
connectWebSocket()
|
||
|
||
/* 启动心跳检测 */
|
||
scheduler.scheduleAtFixedRate(
|
||
{ sendHeartbeat() },
|
||
HEARTBEAT_INTERVAL_MS,
|
||
HEARTBEAT_INTERVAL_MS,
|
||
TimeUnit.MILLISECONDS
|
||
)
|
||
|
||
/* 启动统计输出 */
|
||
scheduler.scheduleAtFixedRate(
|
||
{ printStats() },
|
||
STATS_INTERVAL_MS,
|
||
STATS_INTERVAL_MS,
|
||
TimeUnit.MILLISECONDS
|
||
)
|
||
|
||
/* 启动离线检测（超过30秒无数据视为离线） */
|
||
scheduler.scheduleAtFixedRate(
|
||
{ checkStudentTimeout() },
|
||
10000,
|
||
10000,
|
||
TimeUnit.MILLISECONDS
|
||
)
|
||
}
|
||
|
||
/**
|
||
* 停止笔迹接收
|
||
*/
|
||
fun stop() {
|
||
isRunning.set(false)
|
||
isConnected.set(false)
|
||
|
||
scheduler.shutdown()
|
||
messageExecutor.shutdown()
|
||
|
||
Log.i(TAG, "笔迹接收已停止, 累计接收: ${totalMessagesReceived.get()}条消息, " +
|
||
"${totalPointsReceived.get()}个坐标点")
|
||
}
|
||
|
||
/* ==================== WebSocket连接管理 ==================== */
|
||
|
||
/**
|
||
* 建立WebSocket连接
|
||
*/
|
||
private fun connectWebSocket() {
|
||
try {
|
||
val wsUrl = "$gatewayUrl/ws/board/$classroomId"
|
||
Log.i(TAG, "连接WebSocket: $wsUrl")
|
||
|
||
/* 使用OkHttp WebSocket客户端:
|
||
OkHttpClient.newWebSocket(Request.Builder().url(wsUrl).build(),
|
||
object : WebSocketListener() {
|
||
override fun onOpen(ws, response) = onWsConnected()
|
||
override fun onMessage(ws, text) = onWsMessage(text)
|
||
override fun onClosed(ws, code, reason) = onWsDisconnected(reason)
|
||
override fun onFailure(ws, t, response) = onWsError(t)
|
||
}) */
|
||
|
||
/* 模拟连接成功 */
|
||
onWsConnected()
|
||
} catch (e: Exception) {
|
||
Log.e(TAG, "WebSocket连接失败", e)
|
||
scheduleReconnect()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* WebSocket连接成功回调
|
||
*/
|
||
private fun onWsConnected() {
|
||
isConnected.set(true)
|
||
reconnectDelay = RECONNECT_DELAY_MS
|
||
|
||
Log.i(TAG, "WebSocket已连接, 教室=$classroomId")
|
||
|
||
/* 发送订阅消息 */
|
||
val subscribe = JSONObject().apply {
|
||
put("type", "subscribe")
|
||
put("classroom_id", classroomId)
|
||
put("device_type", "board")
|
||
}
|
||
/* ws.send(subscribe.toString()) */
|
||
|
||
/* 通知监听器 */
|
||
listeners.forEach { it.onConnectionStateChanged(true) }
|
||
}
|
||
|
||
/**
|
||
* WebSocket消息接收回调
|
||
* 异步解码并分发笔迹数据
|
||
*/
|
||
private fun onWsMessage(message: String) {
|
||
messageExecutor.submit {
|
||
try {
|
||
parseAndDispatch(message)
|
||
totalMessagesReceived.incrementAndGet()
|
||
} catch (e: Exception) {
|
||
Log.e(TAG, "消息解析失败: ${e.message}")
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* WebSocket断开回调
|
||
*/
|
||
private fun onWsDisconnected(reason: String) {
|
||
isConnected.set(false)
|
||
Log.w(TAG, "WebSocket已断开: $reason")
|
||
|
||
listeners.forEach { it.onConnectionStateChanged(false) }
|
||
|
||
if (isRunning.get()) {
|
||
scheduleReconnect()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* WebSocket错误回调
|
||
*/
|
||
private fun onWsError(error: Throwable) {
|
||
Log.e(TAG, "WebSocket错误", error)
|
||
isConnected.set(false)
|
||
|
||
if (isRunning.get()) {
|
||
scheduleReconnect()
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 调度重连（指数退避）
|
||
*/
|
||
private fun scheduleReconnect() {
|
||
if (!isRunning.get()) return
|
||
|
||
Log.i(TAG, "将在 ${reconnectDelay}ms 后重连...")
|
||
scheduler.schedule({
|
||
if (isRunning.get() && !isConnected.get()) {
|
||
connectWebSocket()
|
||
}
|
||
}, reconnectDelay, TimeUnit.MILLISECONDS)
|
||
|
||
/* 指数退避增加延迟 */
|
||
reconnectDelay = (reconnectDelay * 1.5).toLong()
|
||
.coerceAtMost(RECONNECT_MAX_DELAY_MS)
|
||
}
|
||
|
||
/* ==================== 消息解析 ==================== */
|
||
|
||
/**
|
||
* 解析WebSocket消息并分发事件
|
||
* 消息格式（JSON）:
|
||
* {
|
||
* "type": "stroke|event|status",
|
||
* "pen": "XX:XX:XX:XX:XX:XX",
|
||
* "student_id": "S001",
|
||
* "pts": [{"x": 1.2, "y": 3.4, "p": 0.5, "t": 123}, ...],
|
||
* "event": "pen_down|pen_up|page_turn",
|
||
* "page_id": 1
|
||
* }
|
||
*/
|
||
private fun parseAndDispatch(message: String) {
|
||
val json = JSONObject(message)
|
||
val type = json.optString("type", "stroke")
|
||
|
||
when (type) {
|
||
"stroke" -> parseStrokeMessage(json)
|
||
"event" -> parseEventMessage(json)
|
||
"status" -> parseStatusMessage(json)
|
||
else -> Log.d(TAG, "未知消息类型: $type")
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析笔迹坐标消息
|
||
*/
|
||
private fun parseStrokeMessage(json: JSONObject) {
|
||
val penId = json.optString("pen", "")
|
||
val studentId = json.optString("student_id", penId)
|
||
val pageId = json.optInt("page_id", 0)
|
||
val ptsArray = json.optJSONArray("pts") ?: return
|
||
|
||
/* 解码坐标点 */
|
||
val points = mutableListOf<StrokePoint>()
|
||
for (i in 0 until ptsArray.length()) {
|
||
val pt = ptsArray.getJSONObject(i)
|
||
points.add(StrokePoint(
|
||
x = pt.optDouble("x", 0.0).toFloat(),
|
||
y = pt.optDouble("y", 0.0).toFloat(),
|
||
pressure = pt.optDouble("p", 0.5).toFloat(),
|
||
timestamp = pt.optLong("t", System.currentTimeMillis())
|
||
))
|
||
}
|
||
|
||
if (points.isEmpty()) return
|
||
|
||
totalPointsReceived.addAndGet(points.size.toLong())
|
||
|
||
/* 更新学生在线状态 */
|
||
if (!onlineStudents.containsKey(penId)) {
|
||
onlineStudents[penId] = studentId
|
||
listeners.forEach { it.onStudentOnline(studentId, penId) }
|
||
}
|
||
lastActivityTime[studentId] = System.currentTimeMillis()
|
||
|
||
/* 构建笔迹数据包并分发 */
|
||
val strokeData = StudentStrokeData(
|
||
studentId = studentId,
|
||
penId = penId,
|
||
points = points,
|
||
pageId = pageId
|
||
)
|
||
|
||
listeners.forEach { it.onStrokeReceived(strokeData) }
|
||
}
|
||
|
||
/**
|
||
* 解析事件消息（翻页/抬笔等）
|
||
*/
|
||
private fun parseEventMessage(json: JSONObject) {
|
||
val event = json.optString("event", "")
|
||
val penId = json.optString("pen", "")
|
||
val studentId = onlineStudents[penId] ?: penId
|
||
|
||
when (event) {
|
||
"page_turn" -> {
|
||
val pageId = json.optInt("page_id", 0)
|
||
listeners.forEach { it.onPageTurn(studentId, pageId) }
|
||
Log.d(TAG, "学生 $studentId 翻页到第 $pageId 页")
|
||
}
|
||
"pen_up" -> {
|
||
Log.d(TAG, "学生 $studentId 抬笔")
|
||
}
|
||
"pen_down" -> {
|
||
Log.d(TAG, "学生 $studentId 落笔")
|
||
}
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 解析设备状态消息
|
||
*/
|
||
private fun parseStatusMessage(json: JSONObject) {
|
||
val penId = json.optString("pen", "")
|
||
val battery = json.optInt("battery", -1)
|
||
if (battery >= 0) {
|
||
Log.d(TAG, "笔 $penId 电量: $battery%")
|
||
}
|
||
}
|
||
|
||
/* ==================== 辅助功能 ==================== */
|
||
|
||
/**
|
||
* 发送心跳
|
||
*/
|
||
private fun sendHeartbeat() {
|
||
if (!isConnected.get()) return
|
||
|
||
val heartbeat = JSONObject().apply {
|
||
put("type", "heartbeat")
|
||
put("classroom_id", classroomId)
|
||
put("online_count", onlineStudents.size)
|
||
put("timestamp", System.currentTimeMillis())
|
||
}
|
||
/* ws.send(heartbeat.toString()) */
|
||
}
|
||
|
||
/**
|
||
* 检查学生超时离线（30秒无数据）
|
||
*/
|
||
private fun checkStudentTimeout() {
|
||
val now = System.currentTimeMillis()
|
||
val timeout = 30000L
|
||
|
||
lastActivityTime.entries.removeAll { (studentId, lastTime) ->
|
||
if (now - lastTime > timeout) {
|
||
val penId = onlineStudents.entries
|
||
.firstOrNull { it.value == studentId }?.key
|
||
penId?.let { onlineStudents.remove(it) }
|
||
|
||
listeners.forEach { it.onStudentOffline(studentId) }
|
||
Log.d(TAG, "学生 $studentId 超时离线")
|
||
true
|
||
} else false
|
||
}
|
||
}
|
||
|
||
/**
|
||
* 输出统计信息
|
||
*/
|
||
private fun printStats() {
|
||
Log.i(TAG, "统计: 在线学生=${onlineStudents.size}, " +
|
||
"累计消息=${totalMessagesReceived.get()}, " +
|
||
"累计坐标点=${totalPointsReceived.get()}, " +
|
||
"已连接=${isConnected.get()}")
|
||
}
|
||
|
||
/**
|
||
* 获取当前在线学生数
|
||
*/
|
||
fun getOnlineStudentCount(): Int = onlineStudents.size
|
||
|
||
/**
|
||
* 获取所有在线学生ID
|
||
*/
|
||
fun getOnlineStudentIds(): Set<String> = onlineStudents.values.toSet()
|
||
}
|