MQTT producer app and IPC context cancellation #54

K3das · 2025-12-23T02:48:02Z

closes #43 and #52

Description

Implements a simple MQTT producer app that reads from GSW and publishes telemetry in the gsw/[packet]/[measurement] topic. This also separates service configuration into a separate docker-compose file to be able to run services in docker without running gsw_service in docker.

How Has This Been Tested?

Fake data generated using ipc_benchmark and mosquitto's mosquitto_sub utility. To test websockets, https://www.hivemq.com/demos/websocket-client/ was used.

TODO

Flags or config files for SHM path, broker address, and topic prefix.
Improved error handling and recovery.
Handle sigint and perform a clean shutdown.
Aggregation
- This will be done in a separate PR given the fairly large scope of the change.

feat: mqtt command line arguments and packet support

Copilot

Pull request overview

This PR implements an MQTT producer application that reads telemetry data from GSW via shared memory and publishes it to an MQTT broker. The changes also refactor the IPC Reader interface to support context-based cancellation for graceful shutdowns, and separate service configurations into a dedicated docker-compose file for more flexible deployment options.

Key changes:

Added context support to the IPC Reader interface enabling cancellation and graceful shutdown
Implemented a new MQTT producer app that publishes telemetry to topic patterns gsw/[packet]/[measurement]
Split Docker Compose configuration to allow running services independently from the main GSW application

Reviewed changes

Copilot reviewed 15 out of 16 changed files in this pull request and generated 15 comments.

Show a summary per file

File	Description
cmd/mqtt_producer/main.go	New MQTT producer app that reads from shared memory and publishes telemetry to MQTT broker with per-measurement topics
lib/ipc/ipc.go	Updated Reader interface to accept context parameter for cancellation support
lib/ipc/shm.go	Implemented context-aware wait() and Read() methods with futex wake on cancellation
lib/ipc/futex.go	Added timeout parameter to futexWait and improved error wrapping with %w
cmd/telem_view/telem_view.go	Updated to pass context.TODO() to new Read() signature
cmd/mem_view/mem_view.go	Updated to pass context.TODO() to new Read() signature
cmd/ipc_benchmark/reader.go	Updated to pass context to Read() for proper cancellation support
cmd/grafana_live/grafana_live.go	Updated to pass context.TODO() to new Read() signature in two locations
compose.yaml	Moved service definitions to separate file and added include statement
compose-services.yaml	New file containing Grafana, InfluxDB, and Mosquitto service definitions
data/mosquitto/mosquitto.conf	Configuration file for Mosquitto broker with anonymous authentication
data/mosquitto/Containerfile	Dockerfile for building custom Mosquitto container with config
cmd/Containerfile	Added mqtt_producer to the list of binaries to build
go.mod	Added paho.mqtt.golang dependency and updated Go version to 1.24.0 with toolchain 1.24.9
go.sum	Updated checksums for new and updated dependencies
README.md	Updated documentation to mention Mosquitto and new compose-services.yaml usage

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

cmd/mqtt_producer/main.go

cmd/mem_view/mem_view.go

cmd/grafana_live/grafana_live.go

data/mosquitto/mosquitto.conf

cmd/mqtt_producer/main.go

cmd/telem_view/telem_view.go

cmd/mqtt_producer/main.go

K3das · 2025-12-27T03:30:40Z

I believe this is a slight performance regression due to the extra context checks.

AarC10 · 2025-12-29T06:06:51Z

I believe this is a slight performance regression due to the extra context checks.

I wouldn't worry too much about it. You have to remember that most of our use cases is testing backplane (~120 pps over Ethernet) and like a packet every few seconds (radio). There will be cases where we may need to deal with higher rates (i.e. DAQ at either 8 khz or 43 khz), but also this is for producing visualization apps where we cannot perceive changes that fast. As long as we can log most of the data meaningfully which is done by the service we should be fine. Optimization is good, but lets not fixate on it unnecessarily (obviously GSW shouldnt be railing the CPU though lol)

AarC10

Looks pretty good! Thanks for consulting and taking my context solution instead of doing polling lol. Im ok with this code, but I think mqtt_producer could see a few improvements down the line

AarC10 · 2025-12-29T06:24:09Z

cmd/mqtt_producer/main.go

+				continue
+			}
+			// qos=0 delivery not guaranteed
+			client.Publish(fmt.Sprintf("%s/%s/%s", *topicPrefix, packet.Name, name), 0, false, jsonStr)


Can we handle the Token return type here?

Nvm read the QOS comment

AarC10 · 2025-12-29T06:32:41Z

cmd/mqtt_producer/main.go

+			val, err := tlm.InterpretMeasurementValue(meas, data[offset:offset+meas.Size])
+			if err != nil {
+				pLog.Printf("error interpreting measurement: %v\n", err)
+				continue
+			}
+			jsonStr, err := json.Marshal(val)
+			if err != nil {
+				pLog.Printf("error marshaling measurement: %v\n", err)
+				continue
+			}
+			// qos=0 delivery not guaranteed
+			client.Publish(fmt.Sprintf("%s/%s/%s", *topicPrefix, packet.Name, name), 0, false, jsonStr)


Im ok with fleshing this out for now, but I think that measurements should be grouped since this would scale poorly. We should avoid having to JSON serializin gand TCP sending to a different topic for each single measurement. This would get expensive fast with more measurements and higher rates

yeahhh- I still want to use different topics for every measurement because it's easiest to consume down the line (with most stuff it will just work ™️. I do want to make better serialization for measurements that isn't running them through JSON- maybe like a generic .String() that will be implemented for every measurement type.

AarC10 · 2025-12-29T06:35:18Z

cmd/mqtt_producer/main.go

+	topicPrefix = flag.String("topic_prefix", "gsw", "mqtt topic prefix")
+)
+
+func main() {


Should document this is a best effort bridge

AarC10 · 2025-12-29T06:38:08Z

cmd/mqtt_producer/main.go

+	client.Disconnect(250)
+}
+
+func packetWriter(ctx context.Context, packet tlm.TelemetryPacket, client mqtt.Client) error {


Another thing I would suggest is adding a way to configure downsampling in the future. If this is mainly for sending to visualization apps that read over MQTT we may not want to be publishing certain things at an insane rate that we can't perceive fast enough or to avoid nuking a computer trying to render millions of data points on a graph

yep! that's the plan for when I end up adding that API.

K3das added 6 commits December 22, 2025 12:01

feat: add primitive mqtt producer implementation

0dba60e

feat: create container for mosquitto and move config

c5ef2df

feat: move docker services into their own config

da8fbf3

feat: add mqtt producer to docker container

4a119e0

feat: context support for ipc reader

60a2229

feat: mqtt command line arguments and packet support

fix: lint

071bea0

K3das requested a review from Copilot December 26, 2025 23:11

Copilot started reviewing on behalf of K3das December 26, 2025 23:12 View session

K3das changed the title ~~MQTT producer app~~ MQTT producer app and IPC context cancellation Dec 26, 2025

Copilot AI reviewed Dec 26, 2025

View reviewed changes

K3das added 3 commits December 26, 2025 15:28

fix: qos, flags, disconnect, error

05cba77

feat: mosquitto websocket listener

69af83b

Merge branch 'main' into feature/mia/mqtt

8c4aa14

K3das marked this pull request as ready for review December 26, 2025 23:43

K3das requested a review from AarC10 December 26, 2025 23:43

K3das linked an issue Dec 26, 2025 that may be closed by this pull request

MQTT producer app #52

Closed

AarC10 approved these changes Dec 29, 2025

View reviewed changes

K3das mentioned this pull request Dec 29, 2025

sampling & aggregation #59

Open

note

360c81f

K3das merged commit c48247e into main Dec 29, 2025
2 checks passed

K3das deleted the feature/mia/mqtt branch December 29, 2025 06:58

MQTT producer app and IPC context cancellation #54

MQTT producer app and IPC context cancellation #54

Uh oh!

Conversation

K3das commented Dec 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Description

How Has This Been Tested?

TODO

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

K3das commented Dec 27, 2025

Uh oh!

AarC10 commented Dec 29, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

AarC10 left a comment

Choose a reason for hiding this comment

Uh oh!

AarC10 Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

AarC10 Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

AarC10 Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

K3das Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

AarC10 Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

AarC10 Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

K3das Dec 29, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

K3das commented Dec 23, 2025 •

edited

Loading

AarC10 commented Dec 29, 2025 •

edited

Loading