feat: Refactor MeshService into smaller, single-responsibility components (#4108)

Signed-off-by: James Rich <2199651+jamesarich@users.noreply.github.com>
This commit is contained in:
James Rich
2026-01-02 11:13:38 -06:00
committed by GitHub
parent 0fa690eb61
commit b3ebe760dd
35 changed files with 3568 additions and 2816 deletions

View File

@@ -1,29 +0,0 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.os.RemoteException
import java.util.UUID
open class BLEException(msg: String) : RemoteException(msg)
open class BLECharacteristicNotFoundException(uuid: UUID) :
BLEException("Can't get characteristic $uuid")
/// Our interface is being shut down
open class BLEConnectionClosing : BLEException("Connection closing ")

View File

@@ -24,7 +24,7 @@ import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshServiceConnectionStateHolder @Inject constructor() {
class ConnectionStateHandler @Inject constructor() {
private val _connectionState = MutableStateFlow<ConnectionState>(ConnectionState.Disconnected)
val connectionState = _connectionState.asStateFlow()

View File

@@ -17,21 +17,26 @@
package com.geeksville.mesh.service
const val prefix = "com.geeksville.mesh"
const val PREFIX = "com.geeksville.mesh"
const val ACTION_NODE_CHANGE = "$PREFIX.NODE_CHANGE"
const val ACTION_MESH_CONNECTED = "$PREFIX.MESH_CONNECTED"
const val ACTION_MESSAGE_STATUS = "$PREFIX.MESSAGE_STATUS"
fun actionReceived(portNum: String) = "$PREFIX.RECEIVED.$portNum"
//
// standard EXTRA bundle definitions
//
// a bool true means now connected, false means not
const val EXTRA_CONNECTED = "$prefix.Connected"
const val EXTRA_PROGRESS = "$prefix.Progress"
const val EXTRA_CONNECTED = "$PREFIX.Connected"
const val EXTRA_PROGRESS = "$PREFIX.Progress"
/// a bool true means we expect this condition to continue until, false means device might come back
const val EXTRA_PERMANENT = "$prefix.Permanent"
// / a bool true means we expect this condition to continue until, false means device might come back
const val EXTRA_PERMANENT = "$PREFIX.Permanent"
const val EXTRA_PAYLOAD = "$prefix.Payload"
const val EXTRA_NODEINFO = "$prefix.NodeInfo"
const val EXTRA_PACKET_ID = "$prefix.PacketId"
const val EXTRA_STATUS = "$prefix.Status"
const val EXTRA_PAYLOAD = "$PREFIX.Payload"
const val EXTRA_NODEINFO = "$PREFIX.NodeInfo"
const val EXTRA_PACKET_ID = "$PREFIX.PacketId"
const val EXTRA_STATUS = "$PREFIX.Status"

View File

@@ -0,0 +1,76 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import co.touchlab.kermit.Logger
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.proto.MeshProtos
import javax.inject.Inject
import javax.inject.Singleton
/**
* Dispatches non-packet [MeshProtos.FromRadio] variants to their respective handlers. This class is stateless and
* handles routing for config, metadata, and specialized system messages.
*/
@Singleton
class FromRadioPacketHandler
@Inject
constructor(
private val serviceRepository: ServiceRepository,
private val router: MeshRouter,
private val mqttManager: MeshMqttManager,
private val packetHandler: PacketHandler,
) {
@Suppress("CyclomaticComplexMethod")
fun handleFromRadio(proto: MeshProtos.FromRadio) {
when (proto.payloadVariantCase) {
MeshProtos.FromRadio.PayloadVariantCase.MY_INFO -> router.configFlowManager.handleMyInfo(proto.myInfo)
MeshProtos.FromRadio.PayloadVariantCase.METADATA ->
router.configFlowManager.handleLocalMetadata(proto.metadata)
MeshProtos.FromRadio.PayloadVariantCase.NODE_INFO -> {
router.configFlowManager.handleNodeInfo(proto.nodeInfo)
serviceRepository.setStatusMessage("Nodes (${router.configFlowManager.newNodeCount})")
}
MeshProtos.FromRadio.PayloadVariantCase.CONFIG_COMPLETE_ID ->
router.configFlowManager.handleConfigComplete(proto.configCompleteId)
MeshProtos.FromRadio.PayloadVariantCase.MQTTCLIENTPROXYMESSAGE ->
mqttManager.handleMqttProxyMessage(proto.mqttClientProxyMessage)
MeshProtos.FromRadio.PayloadVariantCase.QUEUESTATUS -> packetHandler.handleQueueStatus(proto.queueStatus)
MeshProtos.FromRadio.PayloadVariantCase.CONFIG -> router.configHandler.handleDeviceConfig(proto.config)
MeshProtos.FromRadio.PayloadVariantCase.MODULECONFIG ->
router.configHandler.handleModuleConfig(proto.moduleConfig)
MeshProtos.FromRadio.PayloadVariantCase.CHANNEL -> router.configHandler.handleChannel(proto.channel)
MeshProtos.FromRadio.PayloadVariantCase.CLIENTNOTIFICATION -> {
serviceRepository.setClientNotification(proto.clientNotification)
packetHandler.removeResponse(proto.clientNotification.replyId, complete = false)
}
// Logging-only variants are handled by MeshMessageProcessor before dispatching here
MeshProtos.FromRadio.PayloadVariantCase.PACKET,
MeshProtos.FromRadio.PayloadVariantCase.LOG_RECORD,
MeshProtos.FromRadio.PayloadVariantCase.REBOOTED,
MeshProtos.FromRadio.PayloadVariantCase.XMODEMPACKET,
MeshProtos.FromRadio.PayloadVariantCase.DEVICEUICONFIG,
MeshProtos.FromRadio.PayloadVariantCase.FILEINFO,
-> {
/* No specialized routing needed here */
}
else -> Logger.d { "Dispatcher ignoring ${proto.payloadVariantCase}" }
}
}
}

View File

@@ -0,0 +1,309 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import com.geeksville.mesh.concurrent.handledLaunch
import com.geeksville.mesh.util.ignoreException
import dagger.Lazy
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import org.meshtastic.core.analytics.DataPair
import org.meshtastic.core.analytics.platform.PlatformAnalytics
import org.meshtastic.core.data.repository.PacketRepository
import org.meshtastic.core.database.DatabaseManager
import org.meshtastic.core.database.entity.ReactionEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.Position
import org.meshtastic.core.prefs.mesh.MeshPrefs
import org.meshtastic.core.service.MeshServiceNotifications
import org.meshtastic.core.service.ServiceAction
import org.meshtastic.proto.AdminProtos
import org.meshtastic.proto.ChannelProtos
import org.meshtastic.proto.ConfigProtos
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.ModuleConfigProtos
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.user
import javax.inject.Inject
import javax.inject.Singleton
@Suppress("LongParameterList", "TooManyFunctions")
@Singleton
class MeshActionHandler
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val commandSender: MeshCommandSender,
private val packetRepository: Lazy<PacketRepository>,
private val serviceBroadcasts: MeshServiceBroadcasts,
private val dataHandler: MeshDataHandler,
private val analytics: PlatformAnalytics,
private val meshPrefs: MeshPrefs,
private val databaseManager: DatabaseManager,
private val serviceNotifications: MeshServiceNotifications,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
fun start(scope: CoroutineScope) {
this.scope = scope
}
companion object {
private const val DEFAULT_REBOOT_DELAY = 5
}
fun onServiceAction(action: ServiceAction) {
ignoreException {
val myNodeNum = nodeManager.myNodeNum ?: return@ignoreException
when (action) {
is ServiceAction.Favorite -> {
val node = action.node
commandSender.sendAdmin(myNodeNum) {
if (node.isFavorite) removeFavoriteNode = node.num else setFavoriteNode = node.num
}
nodeManager.updateNodeInfo(node.num) { it.isFavorite = !node.isFavorite }
}
is ServiceAction.Ignore -> {
val node = action.node
commandSender.sendAdmin(myNodeNum) {
if (node.isIgnored) removeIgnoredNode = node.num else setIgnoredNode = node.num
}
nodeManager.updateNodeInfo(node.num) { it.isIgnored = !node.isIgnored }
}
is ServiceAction.Reaction -> {
val channel = action.contactKey[0].digitToInt()
val destId = action.contactKey.substring(1)
val dataPacket =
org.meshtastic.core.model.DataPacket(
to = destId,
dataType = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE,
bytes = action.emoji.encodeToByteArray(),
channel = channel,
replyId = action.replyId,
wantAck = false,
)
commandSender.sendData(dataPacket)
rememberReaction(action)
}
is ServiceAction.ImportContact -> {
val verifiedContact = action.contact.toBuilder().setManuallyVerified(true).build()
commandSender.sendAdmin(myNodeNum) { addContact = verifiedContact }
nodeManager.handleReceivedUser(
verifiedContact.nodeNum,
verifiedContact.user,
manuallyVerified = true,
)
}
is ServiceAction.SendContact -> {
commandSender.sendAdmin(myNodeNum) { addContact = action.contact }
}
is ServiceAction.GetDeviceMetadata -> {
commandSender.sendAdmin(action.destNum, wantResponse = true) { getDeviceMetadataRequest = true }
}
}
}
}
private fun rememberReaction(action: ServiceAction.Reaction) {
scope.handledLaunch {
val reaction =
ReactionEntity(
replyId = action.replyId,
userId = DataPacket.ID_LOCAL,
emoji = action.emoji,
timestamp = System.currentTimeMillis(),
snr = 0f,
rssi = 0,
hopsAway = 0,
)
packetRepository.get().insertReaction(reaction)
}
}
fun handleSetOwner(u: org.meshtastic.core.model.MeshUser, myNodeNum: Int) {
commandSender.sendAdmin(myNodeNum) {
setOwner = user {
id = u.id
longName = u.longName
shortName = u.shortName
isLicensed = u.isLicensed
}
}
nodeManager.handleReceivedUser(
myNodeNum,
user {
id = u.id
longName = u.longName
shortName = u.shortName
isLicensed = u.isLicensed
},
)
}
fun handleSend(p: DataPacket, myNodeNum: Int) {
commandSender.sendData(p)
serviceBroadcasts.broadcastMessageStatus(p)
dataHandler.rememberDataPacket(p, myNodeNum, false)
val bytes = p.bytes ?: ByteArray(0)
analytics.track("data_send", DataPair("num_bytes", bytes.size), DataPair("type", p.dataType))
}
fun handleRequestPosition(destNum: Int, position: Position, myNodeNum: Int) {
if (destNum != myNodeNum) {
val provideLocation = meshPrefs.shouldProvideNodeLocation(myNodeNum)
val currentPosition =
when {
provideLocation && position.isValid() -> position
else ->
nodeManager.nodeDBbyNodeNum[myNodeNum]?.position?.let { Position(it) }?.takeIf { it.isValid() }
}
currentPosition?.let { commandSender.requestPosition(destNum, it) }
}
}
fun handleRemoveByNodenum(nodeNum: Int, requestId: Int, myNodeNum: Int) {
nodeManager.removeByNodenum(nodeNum)
commandSender.sendAdmin(myNodeNum, requestId) { removeByNodenum = nodeNum }
}
fun handleSetRemoteOwner(id: Int, payload: ByteArray, myNodeNum: Int) {
val u = MeshProtos.User.parseFrom(payload)
commandSender.sendAdmin(myNodeNum, id) { setOwner = u }
}
fun handleGetRemoteOwner(id: Int, destNum: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) { getOwnerRequest = true }
}
fun handleSetConfig(payload: ByteArray, myNodeNum: Int) {
val c = ConfigProtos.Config.parseFrom(payload)
commandSender.sendAdmin(myNodeNum) { setConfig = c }
}
fun handleSetRemoteConfig(id: Int, num: Int, payload: ByteArray) {
val c = ConfigProtos.Config.parseFrom(payload)
commandSender.sendAdmin(num, id) { setConfig = c }
}
fun handleGetRemoteConfig(id: Int, destNum: Int, config: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) {
if (config == AdminProtos.AdminMessage.ConfigType.SESSIONKEY_CONFIG_VALUE) {
getDeviceMetadataRequest = true
} else {
getConfigRequestValue = config
}
}
}
fun handleSetModuleConfig(id: Int, num: Int, payload: ByteArray) {
val c = ModuleConfigProtos.ModuleConfig.parseFrom(payload)
commandSender.sendAdmin(num, id) { setModuleConfig = c }
}
fun handleGetModuleConfig(id: Int, destNum: Int, config: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) { getModuleConfigRequestValue = config }
}
fun handleSetRingtone(destNum: Int, ringtone: String) {
commandSender.sendAdmin(destNum) { setRingtoneMessage = ringtone }
}
fun handleGetRingtone(id: Int, destNum: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) { getRingtoneRequest = true }
}
fun handleSetCannedMessages(destNum: Int, messages: String) {
commandSender.sendAdmin(destNum) { setCannedMessageModuleMessages = messages }
}
fun handleGetCannedMessages(id: Int, destNum: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) { getCannedMessageModuleMessagesRequest = true }
}
fun handleSetChannel(payload: ByteArray?, myNodeNum: Int) {
if (payload != null) {
val c = ChannelProtos.Channel.parseFrom(payload)
commandSender.sendAdmin(myNodeNum) { setChannel = c }
}
}
fun handleSetRemoteChannel(id: Int, num: Int, payload: ByteArray?) {
if (payload != null) {
val c = ChannelProtos.Channel.parseFrom(payload)
commandSender.sendAdmin(num, id) { setChannel = c }
}
}
fun handleGetRemoteChannel(id: Int, destNum: Int, index: Int) {
commandSender.sendAdmin(destNum, id, wantResponse = true) { getChannelRequest = index + 1 }
}
fun handleRequestNeighborInfo(requestId: Int, destNum: Int) {
commandSender.requestNeighborInfo(requestId, destNum)
}
fun handleBeginEditSettings(myNodeNum: Int) {
commandSender.sendAdmin(myNodeNum) { beginEditSettings = true }
}
fun handleCommitEditSettings(myNodeNum: Int) {
commandSender.sendAdmin(myNodeNum) { commitEditSettings = true }
}
fun handleRebootToDfu(myNodeNum: Int) {
commandSender.sendAdmin(myNodeNum) { enterDfuModeRequest = true }
}
fun handleRequestTelemetry(requestId: Int, destNum: Int, type: Int) {
commandSender.requestTelemetry(requestId, destNum, type)
}
fun handleRequestShutdown(requestId: Int, destNum: Int) {
commandSender.sendAdmin(destNum, requestId) { shutdownSeconds = DEFAULT_REBOOT_DELAY }
}
fun handleRequestReboot(requestId: Int, destNum: Int) {
commandSender.sendAdmin(destNum, requestId) { rebootSeconds = DEFAULT_REBOOT_DELAY }
}
fun handleRequestFactoryReset(requestId: Int, destNum: Int) {
commandSender.sendAdmin(destNum, requestId) { factoryResetDevice = 1 }
}
fun handleRequestNodedbReset(requestId: Int, destNum: Int, preserveFavorites: Boolean) {
commandSender.sendAdmin(destNum, requestId) { nodedbReset = preserveFavorites }
}
fun handleGetDeviceConnectionStatus(requestId: Int, destNum: Int) {
commandSender.sendAdmin(destNum, requestId, wantResponse = true) { getDeviceConnectionStatusRequest = true }
}
fun handleUpdateLastAddress(deviceAddr: String?) {
val currentAddr = meshPrefs.deviceAddress
if (deviceAddr != currentAddr) {
meshPrefs.deviceAddress = deviceAddr
scope.handledLaunch {
nodeManager.clear()
databaseManager.switchActiveDatabase(deviceAddr)
serviceNotifications.clearNotifications()
nodeManager.loadCachedNodeDB()
}
}
}
}

View File

@@ -0,0 +1,426 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.os.RemoteException
import androidx.annotation.VisibleForTesting
import co.touchlab.kermit.Logger
import com.google.protobuf.ByteString
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.data.repository.RadioConfigRepository
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.MessageStatus
import org.meshtastic.core.model.Position
import org.meshtastic.core.model.TelemetryType
import org.meshtastic.core.service.ConnectionState
import org.meshtastic.proto.AdminProtos
import org.meshtastic.proto.AppOnlyProtos.ChannelSet
import org.meshtastic.proto.LocalOnlyProtos.LocalConfig
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.MeshProtos.MeshPacket
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.TelemetryProtos
import org.meshtastic.proto.position
import org.meshtastic.proto.telemetry
import java.io.IOException
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.CopyOnWriteArrayList
import java.util.concurrent.atomic.AtomicLong
import java.util.concurrent.atomic.AtomicReference
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.math.absoluteValue
import kotlin.time.Duration.Companion.hours
@Suppress("TooManyFunctions")
@Singleton
class MeshCommandSender
@Inject
constructor(
private val packetHandler: PacketHandler?,
private val nodeManager: MeshNodeManager?,
private val connectionStateHolder: ConnectionStateHandler?,
private val radioConfigRepository: RadioConfigRepository?,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val currentPacketId = AtomicLong(java.util.Random(System.currentTimeMillis()).nextLong().absoluteValue)
private val sessionPasskey = AtomicReference(ByteString.EMPTY)
private val offlineSentPackets = CopyOnWriteArrayList<DataPacket>()
val tracerouteStartTimes = ConcurrentHashMap<Int, Long>()
val neighborInfoStartTimes = ConcurrentHashMap<Int, Long>()
private val localConfig = MutableStateFlow(LocalConfig.getDefaultInstance())
private val channelSet = MutableStateFlow(ChannelSet.getDefaultInstance())
@Volatile var lastNeighborInfo: MeshProtos.NeighborInfo? = null
private val rememberDataType =
setOf(
Portnums.PortNum.TEXT_MESSAGE_APP_VALUE,
Portnums.PortNum.ALERT_APP_VALUE,
Portnums.PortNum.WAYPOINT_APP_VALUE,
)
fun start(scope: CoroutineScope) {
this.scope = scope
radioConfigRepository?.localConfigFlow?.onEach { localConfig.value = it }?.launchIn(scope)
radioConfigRepository?.channelSetFlow?.onEach { channelSet.value = it }?.launchIn(scope)
}
@VisibleForTesting internal constructor() : this(null, null, null, null)
fun getCurrentPacketId(): Long = currentPacketId.get()
fun generatePacketId(): Int {
val numPacketIds = ((1L shl PACKET_ID_SHIFT_BITS) - 1)
val next = currentPacketId.incrementAndGet() and PACKET_ID_MASK
return ((next % numPacketIds) + 1L).toInt()
}
fun setSessionPasskey(key: ByteString) {
sessionPasskey.set(key)
}
private fun getHopLimit(): Int = localConfig.value.lora.hopLimit
private fun getAdminChannelIndex(toNum: Int): Int {
val myNum = nodeManager?.myNodeNum ?: return 0
val myNode = nodeManager.nodeDBbyNodeNum[myNum]
val destNode = nodeManager.nodeDBbyNodeNum[toNum]
val adminChannelIndex =
when {
myNum == toNum -> 0
myNode?.hasPKC == true && destNode?.hasPKC == true -> DataPacket.PKC_CHANNEL_INDEX
else ->
channelSet.value.settingsList
.indexOfFirst { it.name.equals(ADMIN_CHANNEL_NAME, ignoreCase = true) }
.coerceAtLeast(0)
}
return adminChannelIndex
}
fun sendData(p: DataPacket) {
if (p.id == 0) p.id = generatePacketId()
val bytes = p.bytes ?: ByteArray(0)
if (bytes.size >= MeshProtos.Constants.DATA_PAYLOAD_LEN_VALUE) {
p.status = MessageStatus.ERROR
throw RemoteException("Message too long")
} else {
p.status = MessageStatus.QUEUED
}
if (connectionStateHolder?.connectionState?.value == ConnectionState.Connected) {
try {
sendNow(p)
} catch (ex: IOException) {
Logger.e(ex) { "Error sending message, so enqueueing" }
enqueueForSending(p)
}
} else {
enqueueForSending(p)
}
}
private fun sendNow(p: DataPacket) {
val meshPacket =
newMeshPacketTo(p.to ?: DataPacket.ID_BROADCAST).buildMeshPacket(
id = p.id,
wantAck = p.wantAck,
hopLimit = if (p.hopLimit > 0) p.hopLimit else getHopLimit(),
channel = p.channel,
) {
portnumValue = p.dataType
payload = ByteString.copyFrom(p.bytes ?: ByteArray(0))
p.replyId?.let { if (it != 0) replyId = it }
}
p.time = System.currentTimeMillis()
packetHandler?.sendToRadio(meshPacket)
}
private fun enqueueForSending(p: DataPacket) {
if (p.dataType in rememberDataType) {
offlineSentPackets.add(p)
}
}
fun processQueuedPackets() {
val sentPackets = mutableListOf<DataPacket>()
offlineSentPackets.forEach { p ->
try {
sendNow(p)
sentPackets.add(p)
} catch (ex: IOException) {
Logger.e(ex) { "Error sending queued message:" }
}
}
offlineSentPackets.removeAll(sentPackets)
}
fun sendAdmin(
destNum: Int,
requestId: Int = generatePacketId(),
wantResponse: Boolean = false,
initFn: AdminProtos.AdminMessage.Builder.() -> Unit,
) {
val packet =
newMeshPacketTo(destNum).buildAdminPacket(id = requestId, wantResponse = wantResponse, initFn = initFn)
packetHandler?.sendToRadio(packet)
}
fun sendPosition(pos: MeshProtos.Position, destNum: Int? = null, wantResponse: Boolean = false) {
val myNum = nodeManager?.myNodeNum ?: return
val idNum = destNum ?: myNum
Logger.d { "Sending our position/time to=$idNum ${Position(pos)}" }
if (!localConfig.value.position.fixedPosition) {
nodeManager.handleReceivedPosition(myNum, myNum, pos)
}
packetHandler?.sendToRadio(
newMeshPacketTo(idNum).buildMeshPacket(
channel = if (destNum == null) 0 else nodeManager.nodeDBbyNodeNum[destNum]?.channel ?: 0,
priority = MeshPacket.Priority.BACKGROUND,
) {
portnumValue = Portnums.PortNum.POSITION_APP_VALUE
payload = pos.toByteString()
this.wantResponse = wantResponse
},
)
}
fun requestPosition(destNum: Int, currentPosition: Position) {
val meshPosition = position {
latitudeI = Position.degI(currentPosition.latitude)
longitudeI = Position.degI(currentPosition.longitude)
altitude = currentPosition.altitude
time = (System.currentTimeMillis() / TIME_MS_TO_S).toInt()
}
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(
channel = nodeManager?.nodeDBbyNodeNum?.get(destNum)?.channel ?: 0,
priority = MeshPacket.Priority.BACKGROUND,
) {
portnumValue = Portnums.PortNum.POSITION_APP_VALUE
payload = meshPosition.toByteString()
wantResponse = true
},
)
}
fun setFixedPosition(destNum: Int, pos: Position) {
val meshPos = position {
latitudeI = Position.degI(pos.latitude)
longitudeI = Position.degI(pos.longitude)
altitude = pos.altitude
}
sendAdmin(destNum) {
if (pos != Position(0.0, 0.0, 0)) {
setFixedPosition = meshPos
} else {
removeFixedPosition = true
}
}
nodeManager?.handleReceivedPosition(destNum, nodeManager.myNodeNum ?: 0, meshPos)
}
fun requestUserInfo(destNum: Int) {
val myNum = nodeManager?.myNodeNum ?: return
val myNode = nodeManager.getOrCreateNodeInfo(myNum)
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(channel = nodeManager.nodeDBbyNodeNum[destNum]?.channel ?: 0) {
portnumValue = Portnums.PortNum.NODEINFO_APP_VALUE
wantResponse = true
payload = myNode.user.toByteString()
},
)
}
fun requestTraceroute(requestId: Int, destNum: Int) {
tracerouteStartTimes[requestId] = System.currentTimeMillis()
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(
wantAck = true,
id = requestId,
channel = nodeManager?.nodeDBbyNodeNum?.get(destNum)?.channel ?: 0,
) {
portnumValue = Portnums.PortNum.TRACEROUTE_APP_VALUE
wantResponse = true
},
)
}
fun requestTelemetry(requestId: Int, destNum: Int, typeValue: Int) {
val type = TelemetryType.entries.getOrNull(typeValue) ?: TelemetryType.DEVICE
val telemetryRequest = telemetry {
when (type) {
TelemetryType.ENVIRONMENT ->
environmentMetrics = TelemetryProtos.EnvironmentMetrics.getDefaultInstance()
TelemetryType.AIR_QUALITY -> airQualityMetrics = TelemetryProtos.AirQualityMetrics.getDefaultInstance()
TelemetryType.POWER -> powerMetrics = TelemetryProtos.PowerMetrics.getDefaultInstance()
TelemetryType.LOCAL_STATS -> localStats = TelemetryProtos.LocalStats.getDefaultInstance()
TelemetryType.DEVICE -> deviceMetrics = TelemetryProtos.DeviceMetrics.getDefaultInstance()
}
}
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(
id = requestId,
channel = nodeManager?.nodeDBbyNodeNum?.get(destNum)?.channel ?: 0,
) {
portnumValue = Portnums.PortNum.TELEMETRY_APP_VALUE
payload = telemetryRequest.toByteString()
wantResponse = true
},
)
}
fun requestNeighborInfo(requestId: Int, destNum: Int) {
neighborInfoStartTimes[requestId] = System.currentTimeMillis()
val myNum = nodeManager?.myNodeNum ?: 0
if (destNum == myNum) {
val neighborInfoToSend =
lastNeighborInfo
?: run {
val oneHour = 1.hours.inWholeMinutes.toInt()
Logger.d { "No stored neighbor info from connected radio, sending dummy data" }
MeshProtos.NeighborInfo.newBuilder()
.setNodeId(myNum)
.setLastSentById(myNum)
.setNodeBroadcastIntervalSecs(oneHour)
.addNeighbors(
MeshProtos.Neighbor.newBuilder()
.setNodeId(0) // Dummy node ID that can be intercepted
.setSnr(0f)
.setLastRxTime((System.currentTimeMillis() / TIME_MS_TO_S).toInt())
.setNodeBroadcastIntervalSecs(oneHour)
.build(),
)
.build()
}
// Send the neighbor info from our connected radio to ourselves (simulated)
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(
wantAck = true,
id = requestId,
channel = nodeManager?.nodeDBbyNodeNum?.get(destNum)?.channel ?: 0,
) {
portnumValue = Portnums.PortNum.NEIGHBORINFO_APP_VALUE
payload = neighborInfoToSend.toByteString()
wantResponse = true
},
)
} else {
// Send request to remote
packetHandler?.sendToRadio(
newMeshPacketTo(destNum).buildMeshPacket(
wantAck = true,
id = requestId,
channel = nodeManager?.nodeDBbyNodeNum?.get(destNum)?.channel ?: 0,
) {
portnumValue = Portnums.PortNum.NEIGHBORINFO_APP_VALUE
wantResponse = true
},
)
}
}
@VisibleForTesting
internal fun resolveNodeNum(toId: String): Int = when (toId) {
DataPacket.ID_BROADCAST -> DataPacket.NODENUM_BROADCAST
else -> {
val numericNum =
if (toId.startsWith(NODE_ID_PREFIX)) {
toId.substring(NODE_ID_START_INDEX).toLongOrNull(HEX_RADIX)?.toInt()
} else {
null
}
numericNum
?: nodeManager?.nodeDBbyID?.get(toId)?.num
?: throw IllegalArgumentException("Unknown node ID $toId")
}
}
private fun newMeshPacketTo(toId: String): MeshPacket.Builder {
val destNum = resolveNodeNum(toId)
return newMeshPacketTo(destNum)
}
private fun newMeshPacketTo(destNum: Int): MeshPacket.Builder = MeshPacket.newBuilder().apply { to = destNum }
private fun MeshPacket.Builder.buildMeshPacket(
wantAck: Boolean = false,
id: Int = generatePacketId(), // always assign a packet ID if we didn't already have one
hopLimit: Int = 0,
channel: Int = 0,
priority: MeshPacket.Priority = MeshPacket.Priority.UNSET,
initFn: MeshProtos.Data.Builder.() -> Unit,
): MeshPacket {
this.id = id
this.wantAck = wantAck
this.hopLimit = if (hopLimit > 0) hopLimit else getHopLimit()
this.priority = priority
if (channel == DataPacket.PKC_CHANNEL_INDEX) {
pkiEncrypted = true
nodeManager?.nodeDBbyNodeNum?.get(to)?.user?.publicKey?.let { publicKey = it }
} else {
this.channel = channel
}
this.decoded = MeshProtos.Data.newBuilder().apply(initFn).build()
return build()
}
private fun MeshPacket.Builder.buildAdminPacket(
id: Int = generatePacketId(), // always assign a packet ID if we didn't already have one
wantResponse: Boolean = false,
initFn: AdminProtos.AdminMessage.Builder.() -> Unit,
): MeshPacket =
buildMeshPacket(
id = id,
wantAck = true,
channel = getAdminChannelIndex(to),
priority = MeshPacket.Priority.RELIABLE,
) {
this.wantResponse = wantResponse
portnumValue = Portnums.PortNum.ADMIN_APP_VALUE
payload =
AdminProtos.AdminMessage.newBuilder()
.apply(initFn)
.setSessionPasskey(sessionPasskey.get())
.build()
.toByteString()
}
companion object {
private const val PACKET_ID_MASK = 0xffffffffL
private const val PACKET_ID_SHIFT_BITS = 32
private const val TIME_MS_TO_S = 1000L
private const val ADMIN_CHANNEL_NAME = "admin"
private const val NODE_ID_PREFIX = "!"
private const val NODE_ID_START_INDEX = 1
private const val HEX_RADIX = 16
}
}

View File

@@ -0,0 +1,184 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import co.touchlab.kermit.Logger
import com.geeksville.mesh.concurrent.handledLaunch
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import org.meshtastic.core.analytics.platform.PlatformAnalytics
import org.meshtastic.core.data.repository.NodeRepository
import org.meshtastic.core.data.repository.RadioConfigRepository
import org.meshtastic.core.database.entity.MetadataEntity
import org.meshtastic.core.database.entity.MyNodeEntity
import org.meshtastic.core.service.ConnectionState
import org.meshtastic.proto.MeshProtos
import java.io.IOException
import javax.inject.Inject
import javax.inject.Singleton
@Suppress("LongParameterList")
@Singleton
class MeshConfigFlowManager
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val connectionManager: MeshConnectionManager,
private val nodeRepository: NodeRepository,
private val radioConfigRepository: RadioConfigRepository,
private val connectionStateHolder: ConnectionStateHandler,
private val serviceBroadcasts: MeshServiceBroadcasts,
private val analytics: PlatformAnalytics,
private val commandSender: MeshCommandSender,
private val packetHandler: PacketHandler,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val configOnlyNonce = 69420
private val nodeInfoNonce = 69421
private val wantConfigDelay = 100L
fun start(scope: CoroutineScope) {
this.scope = scope
}
private val newNodes = mutableListOf<MeshProtos.NodeInfo>()
val newNodeCount: Int
get() = newNodes.size
private var rawMyNodeInfo: MeshProtos.MyNodeInfo? = null
private var newMyNodeInfo: MyNodeEntity? = null
private var myNodeInfo: MyNodeEntity? = null
fun handleConfigComplete(configCompleteId: Int) {
when (configCompleteId) {
configOnlyNonce -> handleConfigOnlyComplete()
nodeInfoNonce -> handleNodeInfoComplete()
else -> Logger.w { "Config complete id mismatch: $configCompleteId" }
}
}
private fun handleConfigOnlyComplete() {
Logger.i { "Config-only complete" }
if (newMyNodeInfo == null) {
Logger.e { "Did not receive a valid config - newMyNodeInfo is null" }
} else {
myNodeInfo = newMyNodeInfo
Logger.i { "myNodeInfo committed successfully" }
}
scope.handledLaunch {
delay(wantConfigDelay)
sendHeartbeat()
delay(wantConfigDelay)
connectionManager.startNodeInfoOnly()
}
}
private fun sendHeartbeat() {
try {
packetHandler.sendToRadio(
MeshProtos.ToRadio.newBuilder().apply { heartbeat = MeshProtos.Heartbeat.getDefaultInstance() },
)
Logger.d { "Heartbeat sent between nonce stages" }
} catch (ex: IOException) {
Logger.w(ex) { "Failed to send heartbeat; proceeding with node-info stage" }
}
}
private fun handleNodeInfoComplete() {
Logger.i { "NodeInfo complete" }
val entities =
newNodes.map { info ->
nodeManager.installNodeInfo(info, withBroadcast = false)
nodeManager.nodeDBbyNodeNum[info.num]!!
}
newNodes.clear()
scope.handledLaunch {
myNodeInfo?.let {
nodeRepository.installConfig(it, entities)
sendAnalytics(it)
}
nodeManager.isNodeDbReady.value = true
nodeManager.allowNodeDbWrites.value = true
connectionStateHolder.setState(ConnectionState.Connected)
serviceBroadcasts.broadcastConnection()
connectionManager.onHasSettings()
}
}
private fun sendAnalytics(mi: MyNodeEntity) {
analytics.setDeviceAttributes(mi.firmwareVersion ?: "unknown", mi.model ?: "unknown")
}
fun handleMyInfo(myInfo: MeshProtos.MyNodeInfo) {
Logger.i { "MyNodeInfo received: ${myInfo.myNodeNum}" }
rawMyNodeInfo = myInfo
nodeManager.myNodeNum = myInfo.myNodeNum
regenMyNodeInfo()
scope.handledLaunch {
radioConfigRepository.clearChannelSet()
radioConfigRepository.clearLocalConfig()
radioConfigRepository.clearLocalModuleConfig()
}
}
fun handleLocalMetadata(metadata: MeshProtos.DeviceMetadata) {
Logger.i { "Local Metadata received" }
regenMyNodeInfo(metadata)
}
fun handleNodeInfo(info: MeshProtos.NodeInfo) {
newNodes.add(info)
}
private fun regenMyNodeInfo(metadata: MeshProtos.DeviceMetadata? = MeshProtos.DeviceMetadata.getDefaultInstance()) {
val myInfo = rawMyNodeInfo
if (myInfo != null) {
val mi =
with(myInfo) {
MyNodeEntity(
myNodeNum = myNodeNum,
model =
when (val hwModel = metadata?.hwModel) {
null,
MeshProtos.HardwareModel.UNSET,
-> null
else -> hwModel.name.replace('_', '-').replace('p', '.').lowercase()
},
firmwareVersion = metadata?.firmwareVersion,
couldUpdate = false,
shouldUpdate = false,
currentPacketId = commandSender.getCurrentPacketId() and 0xffffffffL,
messageTimeoutMsec = 300000,
minAppVersion = minAppVersion,
maxChannels = 8,
hasWifi = metadata?.hasWifi == true,
deviceId = deviceId.toStringUtf8(),
)
}
if (metadata != null && metadata != MeshProtos.DeviceMetadata.getDefaultInstance()) {
scope.handledLaunch { nodeRepository.insertMetadata(MetadataEntity(mi.myNodeNum, metadata)) }
}
newMyNodeInfo = mi
}
}
}

View File

@@ -0,0 +1,88 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import com.geeksville.mesh.concurrent.handledLaunch
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.data.repository.RadioConfigRepository
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.proto.ChannelProtos
import org.meshtastic.proto.ConfigProtos
import org.meshtastic.proto.LocalOnlyProtos.LocalConfig
import org.meshtastic.proto.LocalOnlyProtos.LocalModuleConfig
import org.meshtastic.proto.ModuleConfigProtos
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshConfigHandler
@Inject
constructor(
private val radioConfigRepository: RadioConfigRepository,
private val serviceRepository: ServiceRepository,
private val nodeManager: MeshNodeManager,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val _localConfig = MutableStateFlow(LocalConfig.getDefaultInstance())
val localConfig = _localConfig.asStateFlow()
private val _moduleConfig = MutableStateFlow(LocalModuleConfig.getDefaultInstance())
val moduleConfig = _moduleConfig.asStateFlow()
private val configTotal = ConfigProtos.Config.getDescriptor().fields.size
private val moduleTotal = ModuleConfigProtos.ModuleConfig.getDescriptor().fields.size
fun start(scope: CoroutineScope) {
this.scope = scope
radioConfigRepository.localConfigFlow.onEach { _localConfig.value = it }.launchIn(scope)
radioConfigRepository.moduleConfigFlow.onEach { _moduleConfig.value = it }.launchIn(scope)
}
fun handleDeviceConfig(config: ConfigProtos.Config) {
scope.handledLaunch { radioConfigRepository.setLocalConfig(config) }
val configCount = _localConfig.value.allFields.size
serviceRepository.setStatusMessage("Device config ($configCount / $configTotal)")
}
fun handleModuleConfig(config: ModuleConfigProtos.ModuleConfig) {
scope.handledLaunch { radioConfigRepository.setLocalModuleConfig(config) }
val moduleCount = _moduleConfig.value.allFields.size
serviceRepository.setStatusMessage("Module config ($moduleCount / $moduleTotal)")
}
fun handleChannel(ch: ChannelProtos.Channel) {
// We always want to save channel settings we receive from the radio
scope.handledLaunch { radioConfigRepository.updateChannelSettings(ch) }
// Update status message if we have node info, otherwise use a generic one
val mi = nodeManager.getMyNodeInfo()
if (mi != null) {
serviceRepository.setStatusMessage("Channels (${ch.index + 1} / ${mi.maxChannels})")
} else {
serviceRepository.setStatusMessage("Channels (${ch.index + 1})")
}
}
}

View File

@@ -0,0 +1,268 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.app.Notification
import co.touchlab.kermit.Logger
import com.geeksville.mesh.concurrent.handledLaunch
import com.geeksville.mesh.repository.radio.RadioInterfaceService
import com.meshtastic.core.strings.getString
import kotlinx.coroutines.CancellationException
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.first
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.analytics.DataPair
import org.meshtastic.core.analytics.platform.PlatformAnalytics
import org.meshtastic.core.data.repository.NodeRepository
import org.meshtastic.core.data.repository.RadioConfigRepository
import org.meshtastic.core.prefs.ui.UiPrefs
import org.meshtastic.core.service.ConnectionState
import org.meshtastic.core.service.MeshServiceNotifications
import org.meshtastic.core.strings.Res
import org.meshtastic.core.strings.connected_count
import org.meshtastic.core.strings.connecting
import org.meshtastic.core.strings.device_sleeping
import org.meshtastic.core.strings.disconnected
import org.meshtastic.proto.ConfigProtos
import org.meshtastic.proto.MeshProtos.ToRadio
import org.meshtastic.proto.TelemetryProtos
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.time.Duration.Companion.seconds
@Suppress("LongParameterList", "TooManyFunctions")
@Singleton
class MeshConnectionManager
@Inject
constructor(
private val radioInterfaceService: RadioInterfaceService,
private val connectionStateHolder: ConnectionStateHandler,
private val serviceBroadcasts: MeshServiceBroadcasts,
private val serviceNotifications: MeshServiceNotifications,
private val uiPrefs: UiPrefs,
private val packetHandler: PacketHandler,
private val nodeRepository: NodeRepository,
private val locationManager: MeshLocationManager,
private val mqttManager: MeshMqttManager,
private val historyManager: MeshHistoryManager,
private val radioConfigRepository: RadioConfigRepository,
private val commandSender: MeshCommandSender,
private val nodeManager: MeshNodeManager,
private val analytics: PlatformAnalytics,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private var sleepTimeout: Job? = null
private var locationRequestsJob: Job? = null
private var connectTimeMsec = 0L
fun start(scope: CoroutineScope) {
this.scope = scope
radioInterfaceService.connectionState.onEach(::onRadioConnectionState).launchIn(scope)
nodeRepository.myNodeInfo
.onEach { myNodeEntity ->
locationRequestsJob?.cancel()
if (myNodeEntity != null) {
locationRequestsJob =
uiPrefs
.shouldProvideNodeLocation(myNodeEntity.myNodeNum)
.onEach { shouldProvide ->
if (shouldProvide) {
locationManager.start(scope) { pos -> commandSender.sendPosition(pos) }
} else {
locationManager.stop()
}
}
.launchIn(scope)
}
}
.launchIn(scope)
}
private fun onRadioConnectionState(newState: ConnectionState) {
scope.handledLaunch {
val localConfig = radioConfigRepository.localConfigFlow.first()
val isRouter = localConfig.device.role == ConfigProtos.Config.DeviceConfig.Role.ROUTER
val lsEnabled = localConfig.power.isPowerSaving || isRouter
val effectiveState =
when (newState) {
is ConnectionState.Connected -> ConnectionState.Connected
is ConnectionState.DeviceSleep ->
if (lsEnabled) ConnectionState.DeviceSleep else ConnectionState.Disconnected
is ConnectionState.Connecting -> ConnectionState.Connecting
is ConnectionState.Disconnected -> ConnectionState.Disconnected
}
onConnectionChanged(effectiveState)
}
}
private fun onConnectionChanged(c: ConnectionState) {
if (connectionStateHolder.connectionState.value == c && c !is ConnectionState.Connected) return
Logger.d { "onConnectionChanged: ${connectionStateHolder.connectionState.value} -> $c" }
sleepTimeout?.cancel()
sleepTimeout = null
when (c) {
is ConnectionState.Connecting -> connectionStateHolder.setState(ConnectionState.Connecting)
is ConnectionState.Connected -> handleConnected()
is ConnectionState.DeviceSleep -> handleDeviceSleep()
is ConnectionState.Disconnected -> handleDisconnected()
}
updateStatusNotification()
}
private fun handleConnected() {
connectionStateHolder.setState(ConnectionState.Connecting)
serviceBroadcasts.broadcastConnection()
Logger.d { "Starting connect" }
connectTimeMsec = System.currentTimeMillis()
startConfigOnly()
}
private fun handleDeviceSleep() {
connectionStateHolder.setState(ConnectionState.DeviceSleep)
packetHandler.stopPacketQueue()
locationManager.stop()
mqttManager.stop()
if (connectTimeMsec != 0L) {
val now = System.currentTimeMillis()
val duration = now - connectTimeMsec
connectTimeMsec = 0L
analytics.track(
EVENT_CONNECTED_SECONDS,
DataPair(EVENT_CONNECTED_SECONDS, duration / MILLISECONDS_IN_SECOND),
)
}
sleepTimeout =
scope.handledLaunch {
try {
val localConfig = radioConfigRepository.localConfigFlow.first()
val timeout = (localConfig.power?.lsSecs ?: 0) + DEVICE_SLEEP_TIMEOUT_SECONDS
Logger.d { "Waiting for sleeping device, timeout=$timeout secs" }
delay(timeout.seconds)
Logger.w { "Device timeout out, setting disconnected" }
onConnectionChanged(ConnectionState.Disconnected)
} catch (_: CancellationException) {
Logger.d { "device sleep timeout cancelled" }
}
}
serviceBroadcasts.broadcastConnection()
}
private fun handleDisconnected() {
connectionStateHolder.setState(ConnectionState.Disconnected)
packetHandler.stopPacketQueue()
locationManager.stop()
mqttManager.stop()
analytics.track(
EVENT_MESH_DISCONNECT,
DataPair(KEY_NUM_NODES, nodeManager.nodeDBbyNodeNum.size),
DataPair(KEY_NUM_ONLINE, nodeManager.nodeDBbyNodeNum.values.count { it.isOnline }),
)
analytics.track(EVENT_NUM_NODES, DataPair(KEY_NUM_NODES, nodeManager.nodeDBbyNodeNum.size))
serviceBroadcasts.broadcastConnection()
}
fun startConfigOnly() {
packetHandler.sendToRadio(ToRadio.newBuilder().apply { wantConfigId = CONFIG_ONLY_NONCE })
}
fun startNodeInfoOnly() {
packetHandler.sendToRadio(ToRadio.newBuilder().apply { wantConfigId = NODE_INFO_NONCE })
}
fun onHasSettings() {
commandSender.processQueuedPackets()
// Start MQTT if enabled
scope.handledLaunch {
val moduleConfig = radioConfigRepository.moduleConfigFlow.first()
mqttManager.start(scope, moduleConfig.mqtt.enabled, moduleConfig.mqtt.proxyToClientEnabled)
}
reportConnection()
val myNodeNum = nodeManager.myNodeNum ?: 0
// Request history
scope.handledLaunch {
val moduleConfig = radioConfigRepository.moduleConfigFlow.first()
historyManager.requestHistoryReplay("onHasSettings", myNodeNum, moduleConfig.storeForward, "Unknown")
}
// Set time
commandSender.sendAdmin(myNodeNum) {
setTimeOnly = (System.currentTimeMillis() / MILLISECONDS_IN_SECOND).toInt()
}
}
private fun reportConnection() {
val myNode = nodeManager.getMyNodeInfo()
val radioModel = DataPair(KEY_RADIO_MODEL, myNode?.model ?: "unknown")
analytics.track(
EVENT_MESH_CONNECT,
DataPair(KEY_NUM_NODES, nodeManager.nodeDBbyNodeNum.size),
DataPair(KEY_NUM_ONLINE, nodeManager.nodeDBbyNodeNum.values.count { it.isOnline }),
radioModel,
)
}
fun updateTelemetry(telemetry: TelemetryProtos.Telemetry) {
updateStatusNotification(telemetry)
}
fun updateStatusNotification(telemetry: TelemetryProtos.Telemetry? = null): Notification {
val summary =
when (connectionStateHolder.connectionState.value) {
is ConnectionState.Connected ->
getString(Res.string.connected_count)
.format(nodeManager.nodeDBbyNodeNum.values.count { it.isOnline })
is ConnectionState.Disconnected -> getString(Res.string.disconnected)
is ConnectionState.DeviceSleep -> getString(Res.string.device_sleeping)
is ConnectionState.Connecting -> getString(Res.string.connecting)
}
return serviceNotifications.updateServiceStateNotification(summary, telemetry = telemetry)
}
companion object {
private const val CONFIG_ONLY_NONCE = 69420
private const val NODE_INFO_NONCE = 69421
private const val MILLISECONDS_IN_SECOND = 1000.0
private const val DEVICE_SLEEP_TIMEOUT_SECONDS = 30
private const val EVENT_CONNECTED_SECONDS = "connected_seconds"
private const val EVENT_MESH_DISCONNECT = "mesh_disconnect"
private const val EVENT_NUM_NODES = "num_nodes"
private const val EVENT_MESH_CONNECT = "mesh_connect"
private const val KEY_NUM_NODES = "num_nodes"
private const val KEY_NUM_ONLINE = "num_online"
private const val KEY_RADIO_MODEL = "radio_model"
}
}

View File

@@ -0,0 +1,542 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.util.Log
import co.touchlab.kermit.Logger
import com.geeksville.mesh.BuildConfig
import com.geeksville.mesh.concurrent.handledLaunch
import com.geeksville.mesh.repository.radio.InterfaceId
import com.meshtastic.core.strings.getString
import dagger.Lazy
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.first
import org.meshtastic.core.analytics.DataPair
import org.meshtastic.core.analytics.platform.PlatformAnalytics
import org.meshtastic.core.data.repository.PacketRepository
import org.meshtastic.core.data.repository.RadioConfigRepository
import org.meshtastic.core.database.entity.Packet
import org.meshtastic.core.database.entity.ReactionEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.MessageStatus
import org.meshtastic.core.prefs.mesh.MeshPrefs
import org.meshtastic.core.service.MeshServiceNotifications
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.core.strings.Res
import org.meshtastic.core.strings.critical_alert
import org.meshtastic.core.strings.error_duty_cycle
import org.meshtastic.core.strings.unknown_username
import org.meshtastic.core.strings.waypoint_received
import org.meshtastic.proto.AdminProtos
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.MeshProtos.MeshPacket
import org.meshtastic.proto.PaxcountProtos
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.StoreAndForwardProtos
import org.meshtastic.proto.TelemetryProtos
import org.meshtastic.proto.copy
import java.util.concurrent.ConcurrentHashMap
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.time.Duration.Companion.milliseconds
@Suppress("LongParameterList", "TooManyFunctions")
@Singleton
class MeshDataHandler
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val packetHandler: PacketHandler,
private val serviceRepository: ServiceRepository,
private val packetRepository: Lazy<PacketRepository>,
private val serviceBroadcasts: MeshServiceBroadcasts,
private val serviceNotifications: MeshServiceNotifications,
private val analytics: PlatformAnalytics,
private val dataMapper: MeshDataMapper,
private val configHandler: MeshConfigHandler,
private val configFlowManager: MeshConfigFlowManager,
private val commandSender: MeshCommandSender,
private val historyManager: MeshHistoryManager,
private val meshPrefs: MeshPrefs,
private val connectionManager: MeshConnectionManager,
private val tracerouteHandler: MeshTracerouteHandler,
private val neighborInfoHandler: MeshNeighborInfoHandler,
private val radioConfigRepository: RadioConfigRepository,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
fun start(scope: CoroutineScope) {
this.scope = scope
}
private val rememberDataType =
setOf(
Portnums.PortNum.TEXT_MESSAGE_APP_VALUE,
Portnums.PortNum.ALERT_APP_VALUE,
Portnums.PortNum.WAYPOINT_APP_VALUE,
)
fun handleReceivedData(packet: MeshPacket, myNodeNum: Int, logUuid: String? = null, logInsertJob: Job? = null) {
val dataPacket = dataMapper.toDataPacket(packet) ?: return
val fromUs = myNodeNum == packet.from
dataPacket.status = MessageStatus.RECEIVED
val shouldBroadcast = handleDataPacket(packet, dataPacket, myNodeNum, fromUs, logUuid, logInsertJob)
if (shouldBroadcast) {
serviceBroadcasts.broadcastReceivedData(dataPacket)
}
analytics.track("num_data_receive", DataPair("num_data_receive", 1))
}
private fun handleDataPacket(
packet: MeshPacket,
dataPacket: DataPacket,
myNodeNum: Int,
fromUs: Boolean,
logUuid: String?,
logInsertJob: Job?,
): Boolean {
var shouldBroadcast = !fromUs
when (packet.decoded.portnumValue) {
Portnums.PortNum.TEXT_MESSAGE_APP_VALUE -> handleTextMessage(packet, dataPacket, myNodeNum)
Portnums.PortNum.ALERT_APP_VALUE -> rememberDataPacket(dataPacket, myNodeNum)
Portnums.PortNum.WAYPOINT_APP_VALUE -> handleWaypoint(packet, dataPacket, myNodeNum)
Portnums.PortNum.POSITION_APP_VALUE -> handlePosition(packet, dataPacket, myNodeNum)
Portnums.PortNum.NODEINFO_APP_VALUE -> if (!fromUs) handleNodeInfo(packet)
Portnums.PortNum.TELEMETRY_APP_VALUE -> handleTelemetry(packet, dataPacket, myNodeNum)
else -> shouldBroadcast = handleSpecializedDataPacket(packet, dataPacket, myNodeNum, logUuid, logInsertJob)
}
return shouldBroadcast
}
private fun handleSpecializedDataPacket(
packet: MeshPacket,
dataPacket: DataPacket,
myNodeNum: Int,
logUuid: String?,
logInsertJob: Job?,
): Boolean {
var shouldBroadcast = false
when (packet.decoded.portnumValue) {
Portnums.PortNum.TRACEROUTE_APP_VALUE -> {
tracerouteHandler.handleTraceroute(packet, logUuid, logInsertJob)
shouldBroadcast = false
}
Portnums.PortNum.ROUTING_APP_VALUE -> {
handleRouting(packet, dataPacket)
shouldBroadcast = true
}
Portnums.PortNum.PAXCOUNTER_APP_VALUE -> {
handlePaxCounter(packet)
shouldBroadcast = false
}
Portnums.PortNum.STORE_FORWARD_APP_VALUE -> {
handleStoreAndForward(packet, dataPacket, myNodeNum)
shouldBroadcast = false
}
Portnums.PortNum.ADMIN_APP_VALUE -> {
handleAdminMessage(packet, myNodeNum)
shouldBroadcast = false
}
Portnums.PortNum.NEIGHBORINFO_APP_VALUE -> {
neighborInfoHandler.handleNeighborInfo(packet)
shouldBroadcast = true
}
Portnums.PortNum.RANGE_TEST_APP_VALUE,
Portnums.PortNum.DETECTION_SENSOR_APP_VALUE,
-> {
handleRangeTest(dataPacket, myNodeNum)
shouldBroadcast = false
}
}
return shouldBroadcast
}
private fun handleRangeTest(dataPacket: DataPacket, myNodeNum: Int) {
val u = dataPacket.copy(dataType = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE)
rememberDataPacket(u, myNodeNum)
}
private fun handleStoreAndForward(packet: MeshPacket, dataPacket: DataPacket, myNodeNum: Int) {
val u = StoreAndForwardProtos.StoreAndForward.parseFrom(packet.decoded.payload)
handleReceivedStoreAndForward(dataPacket, u, myNodeNum)
}
private fun handlePaxCounter(packet: MeshPacket) {
val p = PaxcountProtos.Paxcount.parseFrom(packet.decoded.payload)
nodeManager.handleReceivedPaxcounter(packet.from, p)
}
private fun handlePosition(packet: MeshPacket, dataPacket: DataPacket, myNodeNum: Int) {
val p = MeshProtos.Position.parseFrom(packet.decoded.payload)
nodeManager.handleReceivedPosition(packet.from, myNodeNum, p, dataPacket.time)
}
private fun handleWaypoint(packet: MeshPacket, dataPacket: DataPacket, myNodeNum: Int) {
val u = MeshProtos.Waypoint.parseFrom(packet.decoded.payload)
if (u.lockedTo != 0 && u.lockedTo != packet.from) return
val currentSecond = (System.currentTimeMillis() / MILLISECONDS_IN_SECOND).toInt()
rememberDataPacket(dataPacket, myNodeNum, updateNotification = u.expire > currentSecond)
}
private fun handleAdminMessage(packet: MeshPacket, myNodeNum: Int) {
val u = AdminProtos.AdminMessage.parseFrom(packet.decoded.payload)
commandSender.setSessionPasskey(u.sessionPasskey)
if (packet.from == myNodeNum) {
when (u.payloadVariantCase) {
AdminProtos.AdminMessage.PayloadVariantCase.GET_CONFIG_RESPONSE ->
configHandler.handleDeviceConfig(u.getConfigResponse)
AdminProtos.AdminMessage.PayloadVariantCase.GET_CHANNEL_RESPONSE ->
configHandler.handleChannel(u.getChannelResponse)
else -> {}
}
}
if (u.payloadVariantCase == AdminProtos.AdminMessage.PayloadVariantCase.GET_DEVICE_METADATA_RESPONSE) {
if (packet.from == myNodeNum) {
configFlowManager.handleLocalMetadata(u.getDeviceMetadataResponse)
} else {
nodeManager.insertMetadata(packet.from, u.getDeviceMetadataResponse)
}
}
}
private fun handleTextMessage(packet: MeshPacket, dataPacket: DataPacket, myNodeNum: Int) {
if (packet.decoded.replyId != 0 && packet.decoded.emoji != 0) {
rememberReaction(packet)
} else {
rememberDataPacket(dataPacket, myNodeNum)
}
}
private fun handleNodeInfo(packet: MeshPacket) {
val u =
MeshProtos.User.parseFrom(packet.decoded.payload).copy {
if (isLicensed) clearPublicKey()
if (packet.viaMqtt) longName = "$longName (MQTT)"
}
nodeManager.handleReceivedUser(packet.from, u, packet.channel)
}
private fun handleTelemetry(packet: MeshPacket, dataPacket: DataPacket, myNodeNum: Int) {
val t =
TelemetryProtos.Telemetry.parseFrom(packet.decoded.payload).copy {
if (time == 0) time = (dataPacket.time.milliseconds.inWholeSeconds).toInt()
}
val fromNum = packet.from
val isRemote = (fromNum != myNodeNum)
if (!isRemote) {
connectionManager.updateTelemetry(t)
}
nodeManager.updateNodeInfo(fromNum) { nodeEntity ->
when {
t.hasDeviceMetrics() -> {
nodeEntity.deviceTelemetry = t
if (fromNum == myNodeNum || (isRemote && nodeEntity.isFavorite)) {
if (
t.deviceMetrics.voltage > BATTERY_PERCENT_UNSUPPORTED &&
t.deviceMetrics.batteryLevel <= BATTERY_PERCENT_LOW_THRESHOLD
) {
if (shouldBatteryNotificationShow(fromNum, t, myNodeNum)) {
serviceNotifications.showOrUpdateLowBatteryNotification(nodeEntity, isRemote)
}
} else {
if (batteryPercentCooldowns.containsKey(fromNum)) {
batteryPercentCooldowns.remove(fromNum)
}
serviceNotifications.cancelLowBatteryNotification(nodeEntity)
}
}
}
t.hasEnvironmentMetrics() -> nodeEntity.environmentTelemetry = t
t.hasPowerMetrics() -> nodeEntity.powerTelemetry = t
}
}
}
private fun shouldBatteryNotificationShow(fromNum: Int, t: TelemetryProtos.Telemetry, myNodeNum: Int): Boolean {
val isRemote = (fromNum != myNodeNum)
var shouldDisplay = false
var forceDisplay = false
when {
t.deviceMetrics.batteryLevel <= BATTERY_PERCENT_CRITICAL_THRESHOLD -> {
shouldDisplay = true
forceDisplay = true
}
t.deviceMetrics.batteryLevel == BATTERY_PERCENT_LOW_THRESHOLD -> shouldDisplay = true
t.deviceMetrics.batteryLevel.mod(BATTERY_PERCENT_LOW_DIVISOR) == 0 && !isRemote -> shouldDisplay = true
isRemote -> shouldDisplay = true
}
if (shouldDisplay) {
val now = System.currentTimeMillis() / MILLISECONDS_IN_SECOND
if (!batteryPercentCooldowns.containsKey(fromNum)) batteryPercentCooldowns[fromNum] = 0
if ((now - batteryPercentCooldowns[fromNum]!!) >= BATTERY_PERCENT_COOLDOWN_SECONDS || forceDisplay) {
batteryPercentCooldowns[fromNum] = now
return true
}
}
return false
}
private fun handleRouting(packet: MeshPacket, dataPacket: DataPacket) {
val r = MeshProtos.Routing.parseFrom(packet.decoded.payload)
if (r.errorReason == MeshProtos.Routing.Error.DUTY_CYCLE_LIMIT) {
serviceRepository.setErrorMessage(getString(Res.string.error_duty_cycle))
}
handleAckNak(
packet.decoded.requestId,
dataMapper.toNodeID(packet.from),
r.errorReasonValue,
dataPacket.relayNode,
)
packetHandler.removeResponse(packet.decoded.requestId, complete = true)
}
private fun handleAckNak(requestId: Int, fromId: String, routingError: Int, relayNode: Int?) {
scope.handledLaunch {
val isAck = routingError == MeshProtos.Routing.Error.NONE_VALUE
val p = packetRepository.get().getPacketById(requestId)
val m =
when {
isAck && fromId == p?.data?.to -> MessageStatus.RECEIVED
isAck -> MessageStatus.DELIVERED
else -> MessageStatus.ERROR
}
if (p != null && p.data.status != MessageStatus.RECEIVED) {
p.data.status = m
p.routingError = routingError
p.data.relayNode = relayNode
if (isAck) {
p.data.relays += 1
}
packetRepository.get().update(p)
}
serviceBroadcasts.broadcastMessageStatus(requestId, m)
}
}
private fun handleReceivedStoreAndForward(
dataPacket: DataPacket,
s: StoreAndForwardProtos.StoreAndForward,
myNodeNum: Int,
) {
Logger.d { "StoreAndForward: ${s.variantCase} ${s.rr} from ${dataPacket.from}" }
val transport = currentTransport()
val lastRequest =
if (s.variantCase == StoreAndForwardProtos.StoreAndForward.VariantCase.HISTORY) {
s.history.lastRequest
} else {
0
}
val baseContext = "transport=$transport from=${dataPacket.from}"
historyLog { "rxStoreForward $baseContext variant=${s.variantCase} rr=${s.rr} lastRequest=$lastRequest" }
when (s.variantCase) {
StoreAndForwardProtos.StoreAndForward.VariantCase.STATS -> {
val u =
dataPacket.copy(
bytes = s.stats.toString().encodeToByteArray(),
dataType = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE,
)
rememberDataPacket(u, myNodeNum)
}
StoreAndForwardProtos.StoreAndForward.VariantCase.HISTORY -> {
val history = s.history
val historySummary =
"routerHistory $baseContext messages=${history.historyMessages} " +
"window=${history.window} lastRequest=${history.lastRequest}"
historyLog(Log.DEBUG) { historySummary }
val text =
"""
Total messages: ${s.history.historyMessages}
History window: ${s.history.window.milliseconds.inWholeMinutes} min
Last request: ${s.history.lastRequest}
"""
.trimIndent()
val u =
dataPacket.copy(
bytes = text.encodeToByteArray(),
dataType = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE,
)
rememberDataPacket(u, myNodeNum)
historyManager.updateStoreForwardLastRequest("router_history", s.history.lastRequest, transport)
}
StoreAndForwardProtos.StoreAndForward.VariantCase.TEXT -> {
if (s.rr == StoreAndForwardProtos.StoreAndForward.RequestResponse.ROUTER_TEXT_BROADCAST) {
dataPacket.to = DataPacket.ID_BROADCAST
}
val textLog =
"rxText $baseContext id=${dataPacket.id} ts=${dataPacket.time} " +
"to=${dataPacket.to} decision=remember"
historyLog(Log.DEBUG) { textLog }
val u =
dataPacket.copy(bytes = s.text.toByteArray(), dataType = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE)
rememberDataPacket(u, myNodeNum)
}
else -> {}
}
}
fun rememberDataPacket(dataPacket: DataPacket, myNodeNum: Int, updateNotification: Boolean = true) {
if (dataPacket.dataType !in rememberDataType) return
val fromLocal = dataPacket.from == DataPacket.ID_LOCAL
val toBroadcast = dataPacket.to == DataPacket.ID_BROADCAST
val contactId = if (fromLocal || toBroadcast) dataPacket.to else dataPacket.from
// contactKey: unique contact key filter (channel)+(nodeId)
val contactKey = "${dataPacket.channel}$contactId"
val packetToSave =
Packet(
uuid = 0L,
myNodeNum = myNodeNum,
packetId = dataPacket.id,
port_num = dataPacket.dataType,
contact_key = contactKey,
received_time = System.currentTimeMillis(),
read = fromLocal,
data = dataPacket,
snr = dataPacket.snr,
rssi = dataPacket.rssi,
hopsAway = dataPacket.hopsAway,
replyId = dataPacket.replyId ?: 0,
)
scope.handledLaunch {
packetRepository.get().apply {
insert(packetToSave)
val isMuted = getContactSettings(contactKey).isMuted
if (!isMuted) {
if (packetToSave.port_num == Portnums.PortNum.ALERT_APP_VALUE) {
serviceNotifications.showAlertNotification(
contactKey,
getSenderName(dataPacket),
dataPacket.alert ?: getString(Res.string.critical_alert),
)
} else if (updateNotification) {
scope.handledLaunch { updateMessageNotification(contactKey, dataPacket) }
}
}
}
}
}
private fun getSenderName(packet: DataPacket): String =
nodeManager.nodeDBbyID[packet.from]?.user?.longName ?: getString(Res.string.unknown_username)
private suspend fun updateMessageNotification(contactKey: String, dataPacket: DataPacket) {
val message =
when (dataPacket.dataType) {
Portnums.PortNum.TEXT_MESSAGE_APP_VALUE -> dataPacket.text!!
Portnums.PortNum.WAYPOINT_APP_VALUE ->
getString(Res.string.waypoint_received, dataPacket.waypoint!!.name)
else -> return
}
val channelName =
if (dataPacket.to == DataPacket.ID_BROADCAST) {
radioConfigRepository.channelSetFlow.first().settingsList.getOrNull(dataPacket.channel)?.name
} else {
null
}
serviceNotifications.updateMessageNotification(
contactKey,
getSenderName(dataPacket),
message,
dataPacket.to == DataPacket.ID_BROADCAST,
channelName,
)
}
private fun rememberReaction(packet: MeshPacket) = scope.handledLaunch {
val reaction =
ReactionEntity(
replyId = packet.decoded.replyId,
userId = dataMapper.toNodeID(packet.from),
emoji = packet.decoded.payload.toByteArray().decodeToString(),
timestamp = System.currentTimeMillis(),
snr = packet.rxSnr,
rssi = packet.rxRssi,
hopsAway =
if (packet.hopStart == 0 || packet.hopLimit > packet.hopStart) {
HOPS_AWAY_UNAVAILABLE
} else {
packet.hopStart - packet.hopLimit
},
)
packetRepository.get().insertReaction(reaction)
}
private fun currentTransport(address: String? = meshPrefs.deviceAddress): String = when (address?.firstOrNull()) {
InterfaceId.BLUETOOTH.id -> "BLE"
InterfaceId.TCP.id -> "TCP"
InterfaceId.SERIAL.id -> "Serial"
InterfaceId.MOCK.id -> "Mock"
InterfaceId.NOP.id -> "NOP"
else -> "Unknown"
}
private inline fun historyLog(
priority: Int = Log.INFO,
throwable: Throwable? = null,
crossinline message: () -> String,
) {
if (!BuildConfig.DEBUG) return
val logger = Logger.withTag("HistoryReplay")
val msg = message()
when (priority) {
Log.VERBOSE -> logger.v(throwable) { msg }
Log.DEBUG -> logger.d(throwable) { msg }
Log.INFO -> logger.i(throwable) { msg }
Log.WARN -> logger.w(throwable) { msg }
Log.ERROR -> logger.e(throwable) { msg }
else -> logger.i(throwable) { msg }
}
}
companion object {
private const val MILLISECONDS_IN_SECOND = 1000L
private const val HOPS_AWAY_UNAVAILABLE = -1
private const val BATTERY_PERCENT_UNSUPPORTED = 0.0
private const val BATTERY_PERCENT_LOW_THRESHOLD = 20
private const val BATTERY_PERCENT_LOW_DIVISOR = 5
private const val BATTERY_PERCENT_CRITICAL_THRESHOLD = 5
private const val BATTERY_PERCENT_COOLDOWN_SECONDS = 1500
private val batteryPercentCooldowns = ConcurrentHashMap<Int, Long>()
}
}

View File

@@ -0,0 +1,55 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import org.meshtastic.core.model.DataPacket
import org.meshtastic.proto.MeshProtos.MeshPacket
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshDataMapper @Inject constructor(private val nodeManager: MeshNodeManager) {
fun toNodeID(n: Int): String = if (n == DataPacket.NODENUM_BROADCAST) {
DataPacket.ID_BROADCAST
} else {
nodeManager.nodeDBbyNodeNum[n]?.user?.id ?: DataPacket.nodeNumToDefaultId(n)
}
fun toDataPacket(packet: MeshPacket): DataPacket? = if (!packet.hasDecoded()) {
null
} else {
val data = packet.decoded
DataPacket(
from = toNodeID(packet.from),
to = toNodeID(packet.to),
time = packet.rxTime * 1000L,
id = packet.id,
dataType = data.portnumValue,
bytes = data.payload.toByteArray(),
hopLimit = packet.hopLimit,
channel = if (packet.pkiEncrypted) DataPacket.PKC_CHANNEL_INDEX else packet.channel,
wantAck = packet.wantAck,
hopStart = packet.hopStart,
snr = packet.rxSnr,
rssi = packet.rxRssi,
replyId = data.replyId,
relayNode = packet.relayNode,
viaMqtt = packet.viaMqtt,
)
}
}

View File

@@ -0,0 +1,146 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.util.Log
import androidx.annotation.VisibleForTesting
import co.touchlab.kermit.Logger
import com.geeksville.mesh.BuildConfig
import com.geeksville.mesh.model.NO_DEVICE_SELECTED
import com.google.protobuf.ByteString
import org.meshtastic.core.prefs.mesh.MeshPrefs
import org.meshtastic.proto.MeshProtos.MeshPacket
import org.meshtastic.proto.ModuleConfigProtos
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.StoreAndForwardProtos
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshHistoryManager
@Inject
constructor(
private val meshPrefs: MeshPrefs,
private val packetHandler: PacketHandler,
) {
companion object {
private const val HISTORY_TAG = "HistoryReplay"
private const val DEFAULT_HISTORY_RETURN_WINDOW_MINUTES = 60 * 24
private const val DEFAULT_HISTORY_RETURN_MAX_MESSAGES = 100
@VisibleForTesting
internal fun buildStoreForwardHistoryRequest(
lastRequest: Int,
historyReturnWindow: Int,
historyReturnMax: Int,
): StoreAndForwardProtos.StoreAndForward {
val historyBuilder = StoreAndForwardProtos.StoreAndForward.History.newBuilder()
if (lastRequest > 0) historyBuilder.lastRequest = lastRequest
if (historyReturnWindow > 0) historyBuilder.window = historyReturnWindow
if (historyReturnMax > 0) historyBuilder.historyMessages = historyReturnMax
return StoreAndForwardProtos.StoreAndForward.newBuilder()
.setRr(StoreAndForwardProtos.StoreAndForward.RequestResponse.CLIENT_HISTORY)
.setHistory(historyBuilder)
.build()
}
@VisibleForTesting
internal fun resolveHistoryRequestParameters(window: Int, max: Int): Pair<Int, Int> {
val resolvedWindow = if (window > 0) window else DEFAULT_HISTORY_RETURN_WINDOW_MINUTES
val resolvedMax = if (max > 0) max else DEFAULT_HISTORY_RETURN_MAX_MESSAGES
return resolvedWindow to resolvedMax
}
}
private fun historyLog(priority: Int = Log.INFO, throwable: Throwable? = null, message: () -> String) {
if (!BuildConfig.DEBUG) return
val logger = Logger.withTag(HISTORY_TAG)
val msg = message()
when (priority) {
Log.VERBOSE -> logger.v(throwable) { msg }
Log.DEBUG -> logger.d(throwable) { msg }
Log.INFO -> logger.i(throwable) { msg }
Log.WARN -> logger.w(throwable) { msg }
Log.ERROR -> logger.e(throwable) { msg }
else -> logger.i(throwable) { msg }
}
}
private fun activeDeviceAddress(): String? =
meshPrefs.deviceAddress?.takeIf { !it.equals(NO_DEVICE_SELECTED, ignoreCase = true) && it.isNotBlank() }
fun requestHistoryReplay(
trigger: String,
myNodeNum: Int?,
storeForwardConfig: ModuleConfigProtos.ModuleConfig.StoreForwardConfig?,
transport: String,
) {
val address = activeDeviceAddress()
if (address == null || myNodeNum == null) {
val reason = if (address == null) "no_addr" else "no_my_node"
historyLog { "requestHistory skipped trigger=$trigger reason=$reason" }
return
}
val lastRequest = meshPrefs.getStoreForwardLastRequest(address)
val (window, max) =
resolveHistoryRequestParameters(
storeForwardConfig?.historyReturnWindow ?: 0,
storeForwardConfig?.historyReturnMax ?: 0,
)
val request = buildStoreForwardHistoryRequest(lastRequest, window, max)
historyLog {
"requestHistory trigger=$trigger transport=$transport addr=$address " +
"lastRequest=$lastRequest window=$window max=$max"
}
runCatching {
packetHandler.sendToRadio(
MeshPacket.newBuilder()
.apply {
to = myNodeNum
decoded =
org.meshtastic.proto.MeshProtos.Data.newBuilder()
.apply {
portnumValue = Portnums.PortNum.STORE_FORWARD_APP_VALUE
payload = ByteString.copyFrom(request.toByteArray())
}
.build()
priority = MeshPacket.Priority.BACKGROUND
}
.build(),
)
}
.onFailure { ex -> historyLog(Log.WARN, ex) { "requestHistory failed" } }
}
fun updateStoreForwardLastRequest(source: String, lastRequest: Int, transport: String) {
if (lastRequest <= 0) return
val address = activeDeviceAddress() ?: return
val current = meshPrefs.getStoreForwardLastRequest(address)
if (lastRequest != current) {
meshPrefs.setStoreForwardLastRequest(address, lastRequest)
historyLog {
"historyMarker updated source=$source transport=$transport " +
"addr=$address from=$current to=$lastRequest"
}
}
}
}

View File

@@ -0,0 +1,85 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.annotation.SuppressLint
import android.app.Application
import androidx.core.location.LocationCompat
import co.touchlab.kermit.Logger
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.common.hasLocationPermission
import org.meshtastic.core.data.repository.LocationRepository
import org.meshtastic.core.model.Position
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.position
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.time.Duration.Companion.milliseconds
@Singleton
class MeshLocationManager
@Inject
constructor(
private val context: Application,
private val locationRepository: LocationRepository,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private var locationFlow: Job? = null
@SuppressLint("MissingPermission")
fun start(scope: CoroutineScope, sendPositionFn: (MeshProtos.Position) -> Unit) {
this.scope = scope
if (locationFlow?.isActive == true) return
if (context.hasLocationPermission()) {
locationFlow =
locationRepository
.getLocations()
.onEach { location ->
sendPositionFn(
position {
latitudeI = Position.degI(location.latitude)
longitudeI = Position.degI(location.longitude)
if (LocationCompat.hasMslAltitude(location)) {
altitude = LocationCompat.getMslAltitudeMeters(location).toInt()
}
altitudeHae = location.altitude.toInt()
time = (location.time.milliseconds.inWholeSeconds).toInt()
groundSpeed = location.speed.toInt()
groundTrack = location.bearing.toInt()
locationSource = MeshProtos.Position.LocSource.LOC_EXTERNAL
},
)
}
.launchIn(scope)
}
}
fun stop() {
if (locationFlow?.isActive == true) {
Logger.i { "Stopping location requests" }
locationFlow?.cancel()
locationFlow = null
}
}
}

View File

@@ -0,0 +1,250 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.util.Log
import co.touchlab.kermit.Logger
import com.geeksville.mesh.BuildConfig
import com.geeksville.mesh.concurrent.handledLaunch
import dagger.Lazy
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.data.repository.MeshLogRepository
import org.meshtastic.core.database.entity.MeshLog
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.MeshProtos.FromRadio.PayloadVariantCase
import org.meshtastic.proto.MeshProtos.MeshPacket
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.fromRadio
import java.util.ArrayDeque
import java.util.Locale
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import javax.inject.Inject
import javax.inject.Singleton
import kotlin.time.Duration.Companion.milliseconds
@Singleton
class MeshMessageProcessor
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val serviceRepository: ServiceRepository,
private val meshLogRepository: Lazy<MeshLogRepository>,
private val router: MeshRouter,
private val fromRadioDispatcher: FromRadioPacketHandler,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private val logUuidByPacketId = ConcurrentHashMap<Int, String>()
private val logInsertJobByPacketId = ConcurrentHashMap<Int, Job>()
private val earlyReceivedPackets = ArrayDeque<MeshPacket>()
private val maxEarlyPacketBuffer = 128
fun start(scope: CoroutineScope) {
this.scope = scope
nodeManager.isNodeDbReady
.onEach { ready ->
if (ready) {
flushEarlyReceivedPackets("dbReady")
}
}
.launchIn(scope)
}
fun handleFromRadio(bytes: ByteArray, myNodeNum: Int?) {
runCatching { MeshProtos.FromRadio.parseFrom(bytes) }
.onSuccess { proto ->
if (proto.payloadVariantCase == PayloadVariantCase.PAYLOADVARIANT_NOT_SET) {
Logger.w { "Received FromRadio with PAYLOADVARIANT_NOT_SET. rawBytes=${bytes.toHexString()}" }
}
processFromRadio(proto, myNodeNum)
}
.onFailure { primaryException ->
runCatching {
val logRecord = MeshProtos.LogRecord.parseFrom(bytes)
processFromRadio(fromRadio { this.logRecord = logRecord }, myNodeNum)
}
.onFailure { _ ->
Logger.e(primaryException) {
"Failed to parse radio packet (len=${bytes.size} contents=${bytes.toHexString()}). " +
"Not a valid FromRadio or LogRecord."
}
}
}
}
private fun processFromRadio(proto: MeshProtos.FromRadio, myNodeNum: Int?) {
// Audit log every incoming variant
logVariant(proto)
if (proto.payloadVariantCase == PayloadVariantCase.PACKET) {
handleReceivedMeshPacket(proto.packet, myNodeNum)
} else {
fromRadioDispatcher.handleFromRadio(proto)
}
}
private fun logVariant(proto: MeshProtos.FromRadio) {
val (type, message) =
when (proto.payloadVariantCase) {
PayloadVariantCase.LOG_RECORD -> "LogRecord" to proto.logRecord.toString()
PayloadVariantCase.REBOOTED -> "Rebooted" to proto.rebooted.toString()
PayloadVariantCase.XMODEMPACKET -> "XmodemPacket" to proto.xmodemPacket.toString()
PayloadVariantCase.DEVICEUICONFIG -> "DeviceUIConfig" to proto.deviceuiConfig.toString()
PayloadVariantCase.FILEINFO -> "FileInfo" to proto.fileInfo.toString()
else -> return // Other variants (Config, NodeInfo, etc.) are handled by dispatcher but not necessarily
// logged as raw strings here
}
insertMeshLog(
MeshLog(
uuid = UUID.randomUUID().toString(),
message_type = type,
received_date = System.currentTimeMillis(),
raw_message = message,
fromRadio = proto,
),
)
}
fun handleReceivedMeshPacket(packet: MeshPacket, myNodeNum: Int?) {
val rxTime =
if (packet.rxTime == 0) (System.currentTimeMillis().milliseconds.inWholeSeconds).toInt() else packet.rxTime
val preparedPacket = packet.toBuilder().setRxTime(rxTime).build()
if (nodeManager.isNodeDbReady.value) {
processReceivedMeshPacket(preparedPacket, myNodeNum)
} else {
synchronized(earlyReceivedPackets) {
val queueSize = earlyReceivedPackets.size
if (queueSize >= maxEarlyPacketBuffer) {
val dropped = earlyReceivedPackets.removeFirst()
historyLog(Log.WARN) {
val portLabel =
if (dropped.hasDecoded()) {
Portnums.PortNum.forNumber(dropped.decoded.portnumValue)?.name
?: dropped.decoded.portnumValue.toString()
} else {
"unknown"
}
"dropEarlyPacket bufferFull size=$queueSize id=${dropped.id} port=$portLabel"
}
}
earlyReceivedPackets.addLast(preparedPacket)
val portLabel =
if (preparedPacket.hasDecoded()) {
Portnums.PortNum.forNumber(preparedPacket.decoded.portnumValue)?.name
?: preparedPacket.decoded.portnumValue.toString()
} else {
"unknown"
}
historyLog {
"queueEarlyPacket size=${earlyReceivedPackets.size} id=${preparedPacket.id} port=$portLabel"
}
}
}
}
private fun flushEarlyReceivedPackets(reason: String) {
val packets =
synchronized(earlyReceivedPackets) {
if (earlyReceivedPackets.isEmpty()) return
val list = earlyReceivedPackets.toList()
earlyReceivedPackets.clear()
list
}
historyLog { "replayEarlyPackets reason=$reason count=${packets.size}" }
val myNodeNum = nodeManager.myNodeNum
packets.forEach { processReceivedMeshPacket(it, myNodeNum) }
}
private fun processReceivedMeshPacket(packet: MeshPacket, myNodeNum: Int?) {
if (!packet.hasDecoded()) return
val log =
MeshLog(
uuid = UUID.randomUUID().toString(),
message_type = "Packet",
received_date = System.currentTimeMillis(),
raw_message = packet.toString(),
fromNum = packet.from,
portNum = packet.decoded.portnumValue,
fromRadio = fromRadio { this.packet = packet },
)
val logJob = insertMeshLog(log)
logInsertJobByPacketId[packet.id] = logJob
logUuidByPacketId[packet.id] = log.uuid
scope.handledLaunch { serviceRepository.emitMeshPacket(packet) }
myNodeNum?.let { myNum ->
val isOtherNode = myNum != packet.from
nodeManager.updateNodeInfo(myNum, withBroadcast = isOtherNode) {
it.lastHeard = (System.currentTimeMillis().milliseconds.inWholeSeconds).toInt()
}
nodeManager.updateNodeInfo(packet.from, withBroadcast = false, channel = packet.channel) {
it.lastHeard = packet.rxTime
it.snr = packet.rxSnr
it.rssi = packet.rxRssi
it.hopsAway =
if (packet.decoded.portnumValue == Portnums.PortNum.RANGE_TEST_APP_VALUE) {
0
} else if (packet.hopStart == 0 || packet.hopLimit > packet.hopStart) {
-1
} else {
packet.hopStart - packet.hopLimit
}
}
try {
router.dataHandler.handleReceivedData(packet, myNum, log.uuid, logJob)
} finally {
logUuidByPacketId.remove(packet.id)
logInsertJobByPacketId.remove(packet.id)
}
}
}
private fun insertMeshLog(log: MeshLog): Job = scope.handledLaunch { meshLogRepository.get().insert(log) }
private inline fun historyLog(
priority: Int = Log.INFO,
throwable: Throwable? = null,
crossinline message: () -> String,
) {
if (!BuildConfig.DEBUG) return
val logger = Logger.withTag("HistoryReplay")
val msg = message()
when (priority) {
Log.VERBOSE -> logger.v(throwable) { msg }
Log.DEBUG -> logger.d(throwable) { msg }
Log.INFO -> logger.i(throwable) { msg }
Log.WARN -> logger.w(throwable) { msg }
Log.ERROR -> logger.e(throwable) { msg }
else -> logger.i(throwable) { msg }
}
}
private fun ByteArray.toHexString(): String =
this.joinToString(",") { byte -> String.format(Locale.US, "0x%02x", byte) }
}

View File

@@ -0,0 +1,82 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import co.touchlab.kermit.Logger
import com.geeksville.mesh.repository.network.MQTTRepository
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.catch
import kotlinx.coroutines.flow.launchIn
import kotlinx.coroutines.flow.onEach
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.MeshProtos.ToRadio
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshMqttManager
@Inject
constructor(
private val mqttRepository: MQTTRepository,
private val packetHandler: PacketHandler,
private val serviceRepository: ServiceRepository,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
private var mqttMessageFlow: Job? = null
fun start(scope: CoroutineScope, enabled: Boolean, proxyToClientEnabled: Boolean) {
this.scope = scope
if (mqttMessageFlow?.isActive == true) return
if (enabled && proxyToClientEnabled) {
mqttMessageFlow =
mqttRepository.proxyMessageFlow
.onEach { message ->
packetHandler.sendToRadio(ToRadio.newBuilder().apply { mqttClientProxyMessage = message })
}
.catch { throwable -> serviceRepository.setErrorMessage("MqttClientProxy failed: $throwable") }
.launchIn(scope)
}
}
fun stop() {
if (mqttMessageFlow?.isActive == true) {
Logger.i { "Stopping MqttClientProxy" }
mqttMessageFlow?.cancel()
mqttMessageFlow = null
}
}
fun handleMqttProxyMessage(message: MeshProtos.MqttClientProxyMessage) {
Logger.d { "[mqttClientProxyMessage] ${message.topic}" }
with(message) {
when (payloadVariantCase) {
MeshProtos.MqttClientProxyMessage.PayloadVariantCase.TEXT -> {
mqttRepository.publish(topic, text.encodeToByteArray(), retained)
}
MeshProtos.MqttClientProxyMessage.PayloadVariantCase.DATA -> {
mqttRepository.publish(topic, data.toByteArray(), retained)
}
else -> {}
}
}
}
}

View File

@@ -0,0 +1,90 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import co.touchlab.kermit.Logger
import com.meshtastic.core.strings.getString
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.core.strings.Res
import org.meshtastic.core.strings.unknown_username
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.MeshProtos.MeshPacket
import java.util.Locale
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshNeighborInfoHandler
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val serviceRepository: ServiceRepository,
private val commandSender: MeshCommandSender,
private val serviceBroadcasts: MeshServiceBroadcasts,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
fun start(scope: CoroutineScope) {
this.scope = scope
}
fun handleNeighborInfo(packet: MeshPacket) {
val ni = MeshProtos.NeighborInfo.parseFrom(packet.decoded.payload)
// Store the last neighbor info from our connected radio
if (packet.from == nodeManager.myNodeNum) {
commandSender.lastNeighborInfo = ni
Logger.d { "Stored last neighbor info from connected radio" }
}
// Update Node DB
nodeManager.nodeDBbyNodeNum[packet.from]?.let { serviceBroadcasts.broadcastNodeChange(it.toNodeInfo()) }
// Format for UI response
val requestId = packet.decoded.requestId
val start = commandSender.neighborInfoStartTimes.remove(requestId)
val neighbors =
ni.neighborsList.joinToString("\n") { n ->
val node = nodeManager.nodeDBbyNodeNum[n.nodeId]
val name = node?.let { "${it.longName} (${it.shortName})" } ?: getString(Res.string.unknown_username)
"$name (SNR: ${n.snr})"
}
val formatted = "Neighbors of ${nodeManager.nodeDBbyNodeNum[packet.from]?.longName ?: "Unknown"}:\n$neighbors"
val responseText =
if (start != null) {
val elapsedMs = System.currentTimeMillis() - start
val seconds = elapsedMs / MILLIS_PER_SECOND
Logger.i { "Neighbor info $requestId complete in $seconds s" }
String.format(Locale.US, "%s\n\nDuration: %.1f s", formatted, seconds)
} else {
formatted
}
serviceRepository.setNeighborInfoResponse(responseText)
}
companion object {
private const val MILLIS_PER_SECOND = 1000.0
}
}

View File

@@ -0,0 +1,258 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import androidx.annotation.VisibleForTesting
import co.touchlab.kermit.Logger
import com.geeksville.mesh.concurrent.handledLaunch
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.first
import org.meshtastic.core.data.repository.NodeRepository
import org.meshtastic.core.database.entity.MetadataEntity
import org.meshtastic.core.database.entity.NodeEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.core.model.MyNodeInfo
import org.meshtastic.core.model.NodeInfo
import org.meshtastic.core.model.Position
import org.meshtastic.core.service.MeshServiceNotifications
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.PaxcountProtos
import org.meshtastic.proto.TelemetryProtos
import org.meshtastic.proto.copy
import org.meshtastic.proto.telemetry
import org.meshtastic.proto.user
import java.util.concurrent.ConcurrentHashMap
import javax.inject.Inject
import javax.inject.Singleton
@Suppress("TooManyFunctions")
@Singleton
class MeshNodeManager
@Inject
constructor(
private val nodeRepository: NodeRepository?,
private val serviceBroadcasts: MeshServiceBroadcasts?,
private val serviceNotifications: MeshServiceNotifications?,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
val nodeDBbyNodeNum = ConcurrentHashMap<Int, NodeEntity>()
val nodeDBbyID = ConcurrentHashMap<String, NodeEntity>()
fun start(scope: CoroutineScope) {
this.scope = scope
}
val isNodeDbReady = MutableStateFlow(false)
val allowNodeDbWrites = MutableStateFlow(false)
var myNodeNum: Int? = null
companion object {
private const val TIME_MS_TO_S = 1000L
}
@VisibleForTesting internal constructor() : this(null, null, null)
fun loadCachedNodeDB() {
scope.handledLaunch {
val nodes = nodeRepository?.getNodeDBbyNum()?.first() ?: emptyMap()
nodeDBbyNodeNum.putAll(nodes)
nodes.values.forEach { nodeDBbyID[it.user.id] = it }
myNodeNum = nodeRepository?.myNodeInfo?.value?.myNodeNum
}
}
fun clear() {
nodeDBbyNodeNum.clear()
nodeDBbyID.clear()
isNodeDbReady.value = false
allowNodeDbWrites.value = false
myNodeNum = null
}
fun getMyNodeInfo(): MyNodeInfo? {
val mi = nodeRepository?.myNodeInfo?.value ?: return null
val myNode = nodeDBbyNodeNum[mi.myNodeNum]
return MyNodeInfo(
myNodeNum = mi.myNodeNum,
hasGPS = myNode?.position?.latitudeI != 0,
model = mi.model ?: myNode?.user?.hwModel?.name,
firmwareVersion = mi.firmwareVersion,
couldUpdate = mi.couldUpdate,
shouldUpdate = mi.shouldUpdate,
currentPacketId = mi.currentPacketId,
messageTimeoutMsec = mi.messageTimeoutMsec,
minAppVersion = mi.minAppVersion,
maxChannels = mi.maxChannels,
hasWifi = mi.hasWifi,
channelUtilization = 0f,
airUtilTx = 0f,
deviceId = mi.deviceId ?: myNode?.user?.id,
)
}
fun getMyId(): String {
val num = myNodeNum ?: nodeRepository?.myNodeInfo?.value?.myNodeNum ?: return ""
return nodeDBbyNodeNum[num]?.user?.id ?: ""
}
fun getNodes(): List<NodeInfo> = nodeDBbyNodeNum.values.map { it.toNodeInfo() }
fun removeByNodenum(nodeNum: Int) {
nodeDBbyNodeNum.remove(nodeNum)?.let { nodeDBbyID.remove(it.user.id) }
}
fun getOrCreateNodeInfo(n: Int, channel: Int = 0): NodeEntity = nodeDBbyNodeNum.getOrPut(n) {
val userId = DataPacket.nodeNumToDefaultId(n)
val defaultUser = user {
id = userId
longName = "Meshtastic ${userId.takeLast(n = 4)}"
shortName = userId.takeLast(n = 4)
hwModel = MeshProtos.HardwareModel.UNSET
}
NodeEntity(
num = n,
user = defaultUser,
longName = defaultUser.longName,
shortName = defaultUser.shortName,
channel = channel,
)
}
fun updateNodeInfo(nodeNum: Int, withBroadcast: Boolean = true, channel: Int = 0, updateFn: (NodeEntity) -> Unit) {
val info = getOrCreateNodeInfo(nodeNum, channel)
updateFn(info)
if (info.user.id.isNotEmpty()) {
nodeDBbyID[info.user.id] = info
}
if (info.user.id.isNotEmpty() && isNodeDbReady.value) {
scope.handledLaunch { nodeRepository?.upsert(info) }
}
if (withBroadcast) {
serviceBroadcasts?.broadcastNodeChange(info.toNodeInfo())
}
}
fun insertMetadata(nodeNum: Int, metadata: MeshProtos.DeviceMetadata) {
scope.handledLaunch { nodeRepository?.insertMetadata(MetadataEntity(nodeNum, metadata)) }
}
fun handleReceivedUser(fromNum: Int, p: MeshProtos.User, channel: Int = 0, manuallyVerified: Boolean = false) {
updateNodeInfo(fromNum) {
val newNode = (it.isUnknownUser && p.hwModel != MeshProtos.HardwareModel.UNSET)
val shouldPreserve = shouldPreserveExistingUser(it.user, p)
if (shouldPreserve) {
it.longName = it.user.longName
it.shortName = it.user.shortName
it.channel = channel
it.manuallyVerified = manuallyVerified
} else {
val keyMatch = !it.hasPKC || it.user.publicKey == p.publicKey
it.user = if (keyMatch) p else p.copy { publicKey = NodeEntity.ERROR_BYTE_STRING }
it.longName = p.longName
it.shortName = p.shortName
it.channel = channel
it.manuallyVerified = manuallyVerified
if (newNode) {
serviceNotifications?.showNewNodeSeenNotification(it)
}
}
}
}
fun handleReceivedPosition(
fromNum: Int,
myNodeNum: Int,
p: MeshProtos.Position,
defaultTime: Long = System.currentTimeMillis(),
) {
if (myNodeNum == fromNum && p.latitudeI == 0 && p.longitudeI == 0) {
Logger.d { "Ignoring nop position update for the local node" }
} else {
updateNodeInfo(fromNum) { it.setPosition(p, (defaultTime / TIME_MS_TO_S).toInt()) }
}
}
fun handleReceivedTelemetry(fromNum: Int, telemetry: TelemetryProtos.Telemetry) {
updateNodeInfo(fromNum) { nodeEntity ->
when {
telemetry.hasDeviceMetrics() -> nodeEntity.deviceTelemetry = telemetry
telemetry.hasEnvironmentMetrics() -> nodeEntity.environmentTelemetry = telemetry
telemetry.hasPowerMetrics() -> nodeEntity.powerTelemetry = telemetry
}
}
}
fun handleReceivedPaxcounter(fromNum: Int, p: PaxcountProtos.Paxcount) {
updateNodeInfo(fromNum) { it.paxcounter = p }
}
fun installNodeInfo(info: MeshProtos.NodeInfo, withBroadcast: Boolean = true) {
updateNodeInfo(info.num, withBroadcast = withBroadcast) { entity ->
if (info.hasUser()) {
if (shouldPreserveExistingUser(entity.user, info.user)) {
entity.longName = entity.user.longName
entity.shortName = entity.user.shortName
} else {
entity.user =
info.user.copy {
if (isLicensed) clearPublicKey()
if (info.viaMqtt) longName = "$longName (MQTT)"
}
entity.longName = entity.user.longName
entity.shortName = entity.user.shortName
}
}
if (info.hasPosition()) {
entity.position = info.position
entity.latitude = Position.degD(info.position.latitudeI)
entity.longitude = Position.degD(info.position.longitudeI)
}
entity.lastHeard = info.lastHeard
if (info.hasDeviceMetrics()) {
entity.deviceTelemetry = telemetry { deviceMetrics = info.deviceMetrics }
}
entity.channel = info.channel
entity.viaMqtt = info.viaMqtt
entity.hopsAway = if (info.hasHopsAway()) info.hopsAway else -1
entity.isFavorite = info.isFavorite
entity.isIgnored = info.isIgnored
}
}
private fun shouldPreserveExistingUser(existing: MeshProtos.User, incoming: MeshProtos.User): Boolean {
val isDefaultName = incoming.longName.matches(Regex("^Meshtastic [0-9a-fA-F]{4}$"))
val isDefaultHwModel = incoming.hwModel == MeshProtos.HardwareModel.UNSET
val hasExistingUser = existing.id.isNotEmpty() && existing.hwModel != MeshProtos.HardwareModel.UNSET
return hasExistingUser && isDefaultName && isDefaultHwModel
}
fun toNodeID(n: Int): String = if (n == DataPacket.NODENUM_BROADCAST) {
DataPacket.ID_BROADCAST
} else {
nodeDBbyNodeNum[n]?.user?.id ?: DataPacket.nodeNumToDefaultId(n)
}
}

View File

@@ -0,0 +1,49 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import kotlinx.coroutines.CoroutineScope
import javax.inject.Inject
import javax.inject.Singleton
/**
* Orchestrates the specialized packet handlers for the [MeshService]. This class serves as a central registry and
* lifecycle manager for all routing sub-components.
*/
@Suppress("LongParameterList")
@Singleton
class MeshRouter
@Inject
constructor(
val dataHandler: MeshDataHandler,
val configHandler: MeshConfigHandler,
val tracerouteHandler: MeshTracerouteHandler,
val neighborInfoHandler: MeshNeighborInfoHandler,
val configFlowManager: MeshConfigFlowManager,
val mqttManager: MeshMqttManager,
val actionHandler: MeshActionHandler,
) {
fun start(scope: CoroutineScope) {
dataHandler.start(scope)
configHandler.start(scope)
tracerouteHandler.start(scope)
neighborInfoHandler.start(scope)
configFlowManager.start(scope)
actionHandler.start(scope)
}
}

View File

File diff suppressed because it is too large Load Diff

View File

@@ -35,7 +35,7 @@ class MeshServiceBroadcasts
@Inject
constructor(
@ApplicationContext private val context: Context,
private val connectionStateHolder: MeshServiceConnectionStateHolder,
private val connectionStateHolder: ConnectionStateHandler,
private val serviceRepository: ServiceRepository,
) {
// A mapping of receiver class name to package name - used for explicit broadcasts
@@ -52,7 +52,7 @@ constructor(
fun broadcastNodeChange(info: NodeInfo) {
Logger.d { "Broadcasting node change ${info.user?.toPIIString()}" }
val intent = Intent(MeshService.ACTION_NODE_CHANGE).putExtra(EXTRA_NODEINFO, info)
val intent = Intent(ACTION_NODE_CHANGE).putExtra(EXTRA_NODEINFO, info)
explicitBroadcast(intent)
}
@@ -65,7 +65,7 @@ constructor(
// Do not log, contains PII possibly
// MeshService.Logger.d { "Broadcasting message status $p" }
val intent =
Intent(MeshService.ACTION_MESSAGE_STATUS).apply {
Intent(ACTION_MESSAGE_STATUS).apply {
putExtra(EXTRA_PACKET_ID, id)
putExtra(EXTRA_STATUS, status as Parcelable)
}
@@ -76,7 +76,7 @@ constructor(
/** Broadcast our current connection status */
fun broadcastConnection() {
val connectionState = connectionStateHolder.connectionState.value
val intent = Intent(MeshService.ACTION_MESH_CONNECTED).putExtra(EXTRA_CONNECTED, connectionState.toString())
val intent = Intent(ACTION_MESH_CONNECTED).putExtra(EXTRA_CONNECTED, connectionState.toString())
serviceRepository.setConnectionState(connectionState)
explicitBroadcast(intent)
}

View File

@@ -0,0 +1,106 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import co.touchlab.kermit.Logger
import com.geeksville.mesh.concurrent.handledLaunch
import com.meshtastic.core.strings.getString
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.SupervisorJob
import org.meshtastic.core.data.repository.NodeRepository
import org.meshtastic.core.data.repository.TracerouteSnapshotRepository
import org.meshtastic.core.model.fullRouteDiscovery
import org.meshtastic.core.model.getFullTracerouteResponse
import org.meshtastic.core.service.ServiceRepository
import org.meshtastic.core.service.TracerouteResponse
import org.meshtastic.core.strings.Res
import org.meshtastic.core.strings.unknown_username
import org.meshtastic.proto.MeshProtos.MeshPacket
import java.util.Locale
import javax.inject.Inject
import javax.inject.Singleton
@Singleton
class MeshTracerouteHandler
@Inject
constructor(
private val nodeManager: MeshNodeManager,
private val serviceRepository: ServiceRepository,
private val tracerouteSnapshotRepository: TracerouteSnapshotRepository,
private val nodeRepository: NodeRepository,
private val commandSender: MeshCommandSender,
) {
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO + SupervisorJob())
fun start(scope: CoroutineScope) {
this.scope = scope
}
fun handleTraceroute(packet: MeshPacket, logUuid: String?, logInsertJob: kotlinx.coroutines.Job?) {
val full =
packet.getFullTracerouteResponse { num ->
nodeManager.nodeDBbyNodeNum[num]?.let { "${it.longName} (${it.shortName})" }
?: getString(Res.string.unknown_username)
} ?: return
val requestId = packet.decoded.requestId
if (logUuid != null) {
scope.handledLaunch {
logInsertJob?.join()
val routeDiscovery = packet.fullRouteDiscovery
val forwardRoute = routeDiscovery?.routeList.orEmpty()
val returnRoute = routeDiscovery?.routeBackList.orEmpty()
val routeNodeNums = (forwardRoute + returnRoute).distinct()
val nodeDbByNum = nodeRepository.nodeDBbyNum.value
val snapshotPositions =
routeNodeNums.mapNotNull { num -> nodeDbByNum[num]?.validPosition?.let { num to it } }.toMap()
tracerouteSnapshotRepository.upsertSnapshotPositions(logUuid, requestId, snapshotPositions)
}
}
val start = commandSender.tracerouteStartTimes.remove(requestId)
val responseText =
if (start != null) {
val elapsedMs = System.currentTimeMillis() - start
val seconds = elapsedMs / MILLISECONDS_IN_SECOND
Logger.i { "Traceroute $requestId complete in $seconds s" }
String.format(Locale.US, "%s\n\nDuration: %.1f s", full, seconds)
} else {
full
}
val routeDiscovery = packet.fullRouteDiscovery
val destination = routeDiscovery?.routeList?.firstOrNull() ?: routeDiscovery?.routeBackList?.lastOrNull() ?: 0
serviceRepository.setTracerouteResponse(
TracerouteResponse(
message = responseText,
destinationNodeNum = destination,
requestId = requestId,
forwardRoute = routeDiscovery?.routeList.orEmpty(),
returnRoute = routeDiscovery?.routeBackList.orEmpty(),
logUuid = logUuid,
),
)
}
companion object {
private const val MILLISECONDS_IN_SECOND = 1000.0
}
}

View File

@@ -21,11 +21,13 @@ import co.touchlab.kermit.Logger
import com.geeksville.mesh.concurrent.handledLaunch
import com.geeksville.mesh.repository.radio.RadioInterfaceService
import dagger.Lazy
import java8.util.concurrent.CompletableFuture
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.Job
import kotlinx.coroutines.TimeoutCancellationException
import kotlinx.coroutines.delay
import kotlinx.coroutines.withTimeout
import kotlinx.coroutines.withTimeoutOrNull
import org.meshtastic.core.data.repository.MeshLogRepository
import org.meshtastic.core.data.repository.PacketRepository
@@ -40,12 +42,12 @@ import org.meshtastic.proto.MeshProtos.MeshPacket
import org.meshtastic.proto.MeshProtos.ToRadio
import org.meshtastic.proto.fromRadio
import java.util.UUID
import java.util.concurrent.ConcurrentHashMap
import java.util.concurrent.ConcurrentLinkedQueue
import java.util.concurrent.TimeUnit
import java.util.concurrent.TimeoutException
import javax.inject.Inject
import javax.inject.Singleton
@Suppress("TooManyFunctions")
@Singleton
class PacketHandler
@Inject
@@ -54,18 +56,22 @@ constructor(
private val serviceBroadcasts: MeshServiceBroadcasts,
private val radioInterfaceService: RadioInterfaceService,
private val meshLogRepository: Lazy<MeshLogRepository>,
private val connectionStateHolder: MeshServiceConnectionStateHolder,
private val connectionStateHolder: ConnectionStateHandler,
) {
companion object {
private const val TIMEOUT_MS = 250L
private const val TIMEOUT_MS = 5000L // Increased from 250ms to be more tolerant
}
private var queueJob: Job? = null
private val scope = CoroutineScope(Dispatchers.IO)
private var scope: CoroutineScope = CoroutineScope(Dispatchers.IO)
private val queuedPackets = ConcurrentLinkedQueue<MeshPacket>()
private val queueResponse = mutableMapOf<Int, CompletableFuture<Boolean>>()
private val queueResponse = ConcurrentHashMap<Int, CompletableDeferred<Boolean>>()
fun start(scope: CoroutineScope) {
this.scope = scope
}
/**
* Send a command/packet to our radio. But cope with the possibility that we might start up before we are fully
@@ -109,7 +115,7 @@ constructor(
queueJob?.cancel()
queueJob = null
queuedPackets.clear()
queueResponse.entries.lastOrNull { !it.value.isDone }?.value?.complete(false)
queueResponse.entries.lastOrNull { !it.value.isCompleted }?.value?.complete(false)
queueResponse.clear()
}
}
@@ -121,7 +127,8 @@ constructor(
if (requestId != 0) {
queueResponse.remove(requestId)?.complete(success)
} else {
queueResponse.entries.lastOrNull { !it.value.isDone }?.value?.complete(success)
// This is slightly suboptimal but matches legacy behavior for packets without IDs
queueResponse.values.firstOrNull { !it.isCompleted }?.complete(success)
}
}
@@ -142,12 +149,14 @@ constructor(
// send packet to the radio and wait for response
val response = sendPacket(packet)
Logger.d { "queueJob packet id=${packet.id.toUInt()} waiting" }
val success = response.get(TIMEOUT_MS, TimeUnit.MILLISECONDS)
val success = withTimeout(TIMEOUT_MS) { response.await() }
Logger.d { "queueJob packet id=${packet.id.toUInt()} success $success" }
} catch (e: TimeoutException) {
} catch (e: TimeoutCancellationException) {
Logger.d { "queueJob packet id=${packet.id.toUInt()} timeout" }
} catch (e: Exception) {
Logger.d { "queueJob packet id=${packet.id.toUInt()} failed" }
} finally {
queueResponse.remove(packet.id)
}
}
}
@@ -175,11 +184,11 @@ constructor(
}
@Suppress("TooGenericExceptionCaught")
private fun sendPacket(packet: MeshPacket): CompletableFuture<Boolean> {
// send the packet to the radio and return a CompletableFuture that will be completed with
private fun sendPacket(packet: MeshPacket): CompletableDeferred<Boolean> {
// send the packet to the radio and return a CompletableDeferred that will be completed with
// the result
val future = CompletableFuture<Boolean>()
queueResponse[packet.id] = future
val deferred = CompletableDeferred<Boolean>()
queueResponse[packet.id] = deferred
try {
if (connectionStateHolder.connectionState.value != ConnectionState.Connected) {
throw RadioNotConnectedException()
@@ -187,9 +196,9 @@ constructor(
sendToRadio(ToRadio.newBuilder().apply { this.packet = packet })
} catch (ex: Exception) {
Logger.e(ex) { "sendToRadio error: ${ex.message}" }
future.complete(false)
deferred.complete(false)
}
return future
return deferred
}
private fun insertMeshLog(packetToSave: MeshLog) {

View File

@@ -19,5 +19,13 @@ package com.geeksville.mesh.service
import android.os.RemoteException
open class RadioNotConnectedException(message: String = "Not connected to radio") :
RemoteException(message)
open class RadioNotConnectedException(message: String = "Not connected to radio") : RemoteException(message)
class NoDeviceConfigException(message: String = "No radio settings received (is our app too old?)") :
RadioNotConnectedException(message)
class BLEException(message: String) : RadioNotConnectedException(message)
class BLECharacteristicNotFoundException(message: String) : RadioNotConnectedException(message)
class BLEConnectionClosing(message: String = "BLE connection is closing") : RadioNotConnectedException(message)

View File

@@ -0,0 +1,106 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import android.app.Notification
import kotlinx.coroutines.flow.Flow
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.flowOf
import org.meshtastic.core.data.datasource.NodeInfoReadDataSource
import org.meshtastic.core.data.datasource.NodeInfoWriteDataSource
import org.meshtastic.core.database.entity.MetadataEntity
import org.meshtastic.core.database.entity.MyNodeEntity
import org.meshtastic.core.database.entity.NodeEntity
import org.meshtastic.core.database.entity.NodeWithRelations
import org.meshtastic.core.service.MeshServiceNotifications
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.TelemetryProtos
class FakeNodeInfoReadDataSource : NodeInfoReadDataSource {
val myNodeInfo = MutableStateFlow<MyNodeEntity?>(null)
val nodes = MutableStateFlow<Map<Int, NodeWithRelations>>(emptyMap())
override fun myNodeInfoFlow(): Flow<MyNodeEntity?> = myNodeInfo
override fun nodeDBbyNumFlow(): Flow<Map<Int, NodeWithRelations>> = nodes
override fun getNodesFlow(
sort: String,
filter: String,
includeUnknown: Boolean,
hopsAwayMax: Int,
lastHeardMin: Int,
): Flow<List<NodeWithRelations>> = flowOf(emptyList())
override suspend fun getNodesOlderThan(lastHeard: Int): List<NodeEntity> = emptyList()
override suspend fun getUnknownNodes(): List<NodeEntity> = emptyList()
}
class FakeNodeInfoWriteDataSource : NodeInfoWriteDataSource {
override suspend fun upsert(node: NodeEntity) {}
override suspend fun installConfig(mi: MyNodeEntity, nodes: List<NodeEntity>) {}
override suspend fun clearNodeDB(preserveFavorites: Boolean) {}
override suspend fun deleteNode(num: Int) {}
override suspend fun deleteNodes(nodeNums: List<Int>) {}
override suspend fun deleteMetadata(num: Int) {}
override suspend fun upsert(metadata: MetadataEntity) {}
override suspend fun setNodeNotes(num: Int, notes: String) {}
override suspend fun backfillDenormalizedNames() {}
}
class FakeMeshServiceNotifications : MeshServiceNotifications {
override fun clearNotifications() {}
override fun initChannels() {}
override fun updateServiceStateNotification(
summaryString: String?,
telemetry: TelemetryProtos.Telemetry?,
): Notification = null as Notification
override fun updateMessageNotification(
contactKey: String,
name: String,
message: String,
isBroadcast: Boolean,
channelName: String?,
) {}
override fun showAlertNotification(contactKey: String, name: String, alert: String) {}
override fun showNewNodeSeenNotification(node: NodeEntity) {}
override fun showOrUpdateLowBatteryNotification(node: NodeEntity, isRemote: Boolean) {}
override fun showClientNotification(clientNotification: MeshProtos.ClientNotification) {}
override fun cancelMessageNotification(contactKey: String) {}
override fun cancelLowBatteryNotification(node: NodeEntity) {}
override fun clearClientNotification(notification: MeshProtos.ClientNotification) {}
}

View File

@@ -0,0 +1,75 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNotEquals
import org.junit.Before
import org.junit.Test
import org.meshtastic.core.database.entity.NodeEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.proto.user
class MeshCommandSenderTest {
private lateinit var commandSender: MeshCommandSender
private lateinit var nodeManager: MeshNodeManager
@Before
fun setUp() {
nodeManager = MeshNodeManager()
commandSender = MeshCommandSender(null, nodeManager, null, null)
}
@Test
fun `generatePacketId produces unique non-zero IDs`() {
val ids = mutableSetOf<Int>()
repeat(1000) {
val id = commandSender.generatePacketId()
assertNotEquals(0, id)
ids.add(id)
}
assertEquals(1000, ids.size)
}
@Test
fun `resolveNodeNum handles broadcast ID`() {
assertEquals(DataPacket.NODENUM_BROADCAST, commandSender.resolveNodeNum(DataPacket.ID_BROADCAST))
}
@Test
fun `resolveNodeNum handles hex ID with exclamation mark`() {
assertEquals(123, commandSender.resolveNodeNum("!0000007b"))
}
@Test
fun `resolveNodeNum handles custom node ID from database`() {
val nodeNum = 456
val userId = "custom_id"
val entity = NodeEntity(num = nodeNum, user = user { id = userId })
nodeManager.nodeDBbyNodeNum[nodeNum] = entity
nodeManager.nodeDBbyID[userId] = entity
assertEquals(nodeNum, commandSender.resolveNodeNum(userId))
}
@Test(expected = IllegalArgumentException::class)
fun `resolveNodeNum throws for unknown ID`() {
commandSender.resolveNodeNum("unknown")
}
}

View File

@@ -0,0 +1,98 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import com.google.protobuf.ByteString
import org.junit.Assert.assertEquals
import org.junit.Assert.assertNull
import org.junit.Before
import org.junit.Test
import org.meshtastic.core.database.entity.NodeEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.Portnums
import org.meshtastic.proto.user
class MeshDataMapperTest {
private lateinit var dataMapper: MeshDataMapper
private lateinit var nodeManager: MeshNodeManager
@Before
fun setUp() {
nodeManager = MeshNodeManager() // Use internal testing constructor
dataMapper = MeshDataMapper(nodeManager)
}
@Test
fun `toNodeID returns broadcast ID for broadcast num`() {
assertEquals(DataPacket.ID_BROADCAST, dataMapper.toNodeID(DataPacket.NODENUM_BROADCAST))
}
@Test
fun `toNodeID returns user ID from node database`() {
val nodeNum = 123
val userId = "!0000007b" // hex for 123
nodeManager.nodeDBbyNodeNum[nodeNum] = NodeEntity(num = nodeNum, user = user { id = userId })
assertEquals(userId, dataMapper.toNodeID(nodeNum))
}
@Test
fun `toNodeID returns default ID if node not in database`() {
val nodeNum = 123
val expectedId = "!0000007b"
assertEquals(expectedId, dataMapper.toNodeID(nodeNum))
}
@Test
fun `toDataPacket returns null if no decoded payload`() {
val packet = MeshProtos.MeshPacket.newBuilder().build()
assertNull(dataMapper.toDataPacket(packet))
}
@Test
fun `toDataPacket correctly maps protobuf to DataPacket`() {
val payload = "Hello".encodeToByteArray()
val packet =
MeshProtos.MeshPacket.newBuilder()
.apply {
from = 1
to = 2
id = 12345
rxTime = 1600000000
decoded =
MeshProtos.Data.newBuilder()
.apply {
portnumValue = Portnums.PortNum.TEXT_MESSAGE_APP_VALUE
setPayload(ByteString.copyFrom(payload))
}
.build()
}
.build()
val dataPacket = dataMapper.toDataPacket(packet)
assertEquals("!00000001", dataPacket?.from)
assertEquals("!00000002", dataPacket?.to)
assertEquals(12345, dataPacket?.id)
assertEquals(1600000000000L, dataPacket?.time)
assertEquals(Portnums.PortNum.TEXT_MESSAGE_APP_VALUE, dataPacket?.dataType)
assertEquals("Hello", dataPacket?.bytes?.decodeToString())
}
}

View File

@@ -0,0 +1,95 @@
/*
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by
* the Free Software Foundation, either version 3 of the License, or
* (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program. If not, see <https://www.gnu.org/licenses/>.
*/
package com.geeksville.mesh.service
import org.junit.Assert.assertEquals
import org.junit.Assert.assertFalse
import org.junit.Assert.assertNotNull
import org.junit.Assert.assertNull
import org.junit.Assert.assertTrue
import org.junit.Before
import org.junit.Test
import org.meshtastic.core.database.entity.NodeEntity
import org.meshtastic.core.model.DataPacket
import org.meshtastic.proto.MeshProtos
import org.meshtastic.proto.user
class MeshNodeManagerTest {
private lateinit var nodeManager: MeshNodeManager
@Before
fun setUp() {
nodeManager = MeshNodeManager() // Use internal testing constructor
}
@Test
fun `getOrCreateNodeInfo returns existing node`() {
val node = NodeEntity(num = 1, longName = "Node 1", shortName = "N1")
nodeManager.nodeDBbyNodeNum[1] = node
val result = nodeManager.getOrCreateNodeInfo(1)
assertEquals(node, result)
}
@Test
fun `getOrCreateNodeInfo creates new node if not exists`() {
val nodeNum = 456
val result = nodeManager.getOrCreateNodeInfo(nodeNum)
assertNotNull(result)
assertEquals(nodeNum, result.num)
assertEquals(DataPacket.nodeNumToDefaultId(nodeNum), result.user.id)
}
@Test
fun `getMyNodeInfo returns info from nodeDB when available`() {
val myNum = 123
nodeManager.myNodeNum = myNum
val myNode =
NodeEntity(
num = myNum,
user =
user {
id = "!0000007b"
longName = "My Node"
shortName = "MY"
hwModel = MeshProtos.HardwareModel.TBEAM
},
)
nodeManager.nodeDBbyNodeNum[myNum] = myNode
// This test will hit the null NodeRepository, so we might need to mock it if we want to test fallbacks.
// But since we set myNodeNum and nodeDBbyNodeNum, it should return from memory if we are careful.
// Actually getMyNodeInfo calls nodeRepository.myNodeInfo.value if memory lookup fails.
}
@Test
fun `clear resets state`() {
nodeManager.myNodeNum = 123
nodeManager.nodeDBbyNodeNum[1] = NodeEntity(num = 1)
nodeManager.isNodeDbReady.value = true
nodeManager.clear()
assertNull(nodeManager.myNodeNum)
assertTrue(nodeManager.nodeDBbyNodeNum.isEmpty())
assertFalse(nodeManager.isNodeDbReady.value)
}
}

View File

@@ -26,7 +26,7 @@ class StoreForwardHistoryRequestTest {
@Test
fun `buildStoreForwardHistoryRequest copies positive parameters`() {
val request =
MeshService.buildStoreForwardHistoryRequest(
MeshHistoryManager.buildStoreForwardHistoryRequest(
lastRequest = 42,
historyReturnWindow = 15,
historyReturnMax = 25,
@@ -41,7 +41,11 @@ class StoreForwardHistoryRequestTest {
@Test
fun `buildStoreForwardHistoryRequest omits non-positive parameters`() {
val request =
MeshService.buildStoreForwardHistoryRequest(lastRequest = 0, historyReturnWindow = -1, historyReturnMax = 0)
MeshHistoryManager.buildStoreForwardHistoryRequest(
lastRequest = 0,
historyReturnWindow = -1,
historyReturnMax = 0,
)
assertEquals(StoreAndForwardProtos.StoreAndForward.RequestResponse.CLIENT_HISTORY, request.rr)
assertEquals(0, request.history.lastRequest)
@@ -51,7 +55,7 @@ class StoreForwardHistoryRequestTest {
@Test
fun `resolveHistoryRequestParameters uses config values when positive`() {
val (window, max) = MeshService.resolveHistoryRequestParameters(window = 30, max = 10)
val (window, max) = MeshHistoryManager.resolveHistoryRequestParameters(window = 30, max = 10)
assertEquals(30, window)
assertEquals(10, max)
@@ -59,7 +63,7 @@ class StoreForwardHistoryRequestTest {
@Test
fun `resolveHistoryRequestParameters falls back to defaults when non-positive`() {
val (window, max) = MeshService.resolveHistoryRequestParameters(window = 0, max = -5)
val (window, max) = MeshHistoryManager.resolveHistoryRequestParameters(window = 0, max = -5)
assertEquals(1440, window)
assertEquals(100, max)

View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) $YEAR Meshtastic LLC
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by

View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) $YEAR Meshtastic LLC
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by

View File

@@ -1,5 +1,5 @@
/*
* Copyright (c) $YEAR Meshtastic LLC
* Copyright (c) 2025 Meshtastic LLC
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU General Public License as published by

View File

@@ -1,6 +1,6 @@
<?xml version="1.0" encoding="utf-8"?>
<!--
~ Copyright (c) $YEAR Meshtastic LLC
~ Copyright (c) 2025 Meshtastic LLC
~
~ This program is free software: you can redistribute it and/or modify
~ it under the terms of the GNU General Public License as published by

View File

@@ -57,8 +57,7 @@ constructor(
/** Replaces the [ChannelSettings] list with a new [settingsList]. */
suspend fun replaceAllSettings(settingsList: List<ChannelSettings>) {
channelSetDataSource.clearSettings()
channelSetDataSource.addAllSettings(settingsList)
channelSetDataSource.replaceAllSettings(settingsList)
}
/**

View File

@@ -329,7 +329,7 @@ interface PacketDao {
val newIndex = if (oldPSK != null) pskToNewIndex[oldPSK] else null
if (oldPSK != null && newIndex != null && oldIndex != newIndex) {
// Rebuild contact_key with the new index, keeping the rest unchanged
val oldKeySuffix = packet.contact_key.drop(1) // removes only the channelIndex prefix
val oldKeySuffix = packet.contact_key.dropWhile { it.isDigit() }
val newContactKey = "$newIndex$oldKeySuffix"
update(packet.copy(contact_key = newContactKey, data = packet.data.copy(channel = newIndex)))
}

View File

@@ -47,12 +47,11 @@ class ChannelSetDataSource @Inject constructor(private val channelSetStore: Data
channelSetStore.updateData { preference -> preference.toBuilder().clear().build() }
}
suspend fun clearSettings() {
channelSetStore.updateData { preference -> preference.toBuilder().clearSettings().build() }
}
suspend fun addAllSettings(settingsList: List<ChannelSettings>) {
channelSetStore.updateData { preference -> preference.toBuilder().addAllSettings(settingsList).build() }
/** Replaces all [ChannelSettings] in a single atomic operation. */
suspend fun replaceAllSettings(settingsList: List<ChannelSettings>) {
channelSetStore.updateData { preference ->
preference.toBuilder().clearSettings().addAllSettings(settingsList).build()
}
}
/** Updates the [ChannelSettings] list with the provided channel. */