Skip to content

Add team broker mqtt agent support #193

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
Sep 22, 2025
Merged

Add team broker mqtt agent support #193

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
08c5456
Add team broker mqtt agent support
hardillb Sep 10, 2025
ec3d285
fix lint
hardillb Sep 10, 2025
43e8876
Apply suggestions from code review (remove console.log)
hardillb Sep 16, 2025
1653928
Ensure MQTT schema agents are stopped/restarted
hardillb Sep 17, 2025
e389424
Merge branch 'team-broker-schema' of https://github.com/FlowFuse/driv…
hardillb Sep 17, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 17 additions & 6 deletions localfs.js
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -228,12 +228,17 @@ async function getMQTTAgentList (driver) {
const agents = await driver._app.db.models.BrokerCredentials.findAll({
include: [{ model: driver._app.db.models.Team }]
})
const teamBrokerAgents = await driver._app.db.models.TeamBrokerAgent.findAll({
include: [{ model: driver._app.db.models.Team }]
})

agents.concat(teamBrokerAgents)
agents.forEach(async (agent) => {
if (agent.Team && agent.settings.port) {
driver._usedAgentPorts.add(agent.settings.port)
}
})

return agents
}

Expand Down Expand Up @@ -306,7 +311,9 @@ async function checkExistingMQTTAgents (driver) {
}

async function launchMQTTAgent (broker, driver) {
logger.info(`[localfs] Starting MQTT Schema agent ${broker.hashid} for ${broker.Team.hashid}`)
const agent = broker.constructor.name === 'TeamBrokerAgent'

logger.info(`[localfs] Starting MQTT Schema agent ${agent ? 'team-broker' : broker.hashid} for ${broker.Team.hashid}`)
const agentSettings = broker.settings
agentSettings.port = agentSettings.port || getNextFreePort(driver._usedAgentPorts, initialAgentPortNumber)

Expand All @@ -315,26 +322,29 @@ async function launchMQTTAgent (broker, driver) {

env.FORGE_TEAM_TOKEN = token
env.FORGE_URL = driver._app.config.base_url
env.FORGE_BROKER_ID = broker.hashid
env.FORGE_BROKER_ID = agent ? 'team-broker' : broker.hashid
env.FORGE_TEAM_ID = broker.Team.hashid
env.FORGE_PORT = agentSettings.port
if (agent) {
env.FORGE_TIMEOUT = 24
}

if (driver._app.config.node_path) {
env.PATH = process.env.PATH + path.delimiter + driver._app.config.node_path
} else {
env.PATH = process.env.PATH
}

const out = openSync(path.join(driver._rootDir, `/${broker.hashid}-out.log`), 'a')
const err = openSync(path.join(driver._rootDir, `/${broker.hashid}-out.log`), 'a')
const out = openSync(path.join(driver._rootDir, `/${agent ? 'team-broker-' + broker.Team.hashid : broker.hashid}-out.log`), 'a')
const err = openSync(path.join(driver._rootDir, `/${agent ? 'team-broker-' + broker.Team.hashid : broker.hashid}-out.log`), 'a')

fileHandles[broker.hashid] = {
out,
err
}

const processOptions = {
detached: true,
detached: false,
windowsHide: true,
stdio: ['ignore', out, err],
env,
Expand Down Expand Up @@ -760,9 +770,10 @@ module.exports = {
launchMQTTAgent(broker, this)
},
stopBrokerAgent: async (broker) => {
const agent = broker.constructor.name === 'TeamBrokerAgent'
const pid = broker.settings?.pid
const port = broker.settings?.port
logger.info(`Stopping MQTT Schema Agent ${broker.hashid}`)
logger.info(`Stopping MQTT Schema Agent ${agent ? 'team-broker' : broker.hashid} for ${broker.Team.hashid}`)
if (pid) {
try {
if (process.platform === 'win32') {
Expand Down