Ryan
fb30e8e884
Move lesson source into the teaching structure, align the course path with the learner's Go and DevOps experience, and add the interactive lesson guides. Add advanced DevOps lifecycle study as a separate section. Bound MQTT message work and drain accepted jobs during shutdown; align the backend implementation with the revised course through Section 6.
116 lines
3.3 KiB
Go
116 lines
3.3 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"fmt"
|
|
"log"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"syscall"
|
|
"time"
|
|
|
|
"example.com/mqtt5-learning/internal/config"
|
|
"example.com/mqtt5-learning/internal/contracts"
|
|
"example.com/mqtt5-learning/internal/deviceview"
|
|
"example.com/mqtt5-learning/internal/dispatch"
|
|
"example.com/mqtt5-learning/internal/mqtt"
|
|
)
|
|
|
|
func main() {
|
|
if err := run(); err != nil {
|
|
log.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func run() error {
|
|
cfg, err := config.Load(os.Args[1:])
|
|
if err != nil {
|
|
return fmt.Errorf("failed to load config: %w", err)
|
|
}
|
|
|
|
lifecycleCtx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
transportCtx, stopTransport := context.WithCancel(context.Background())
|
|
defer stopTransport()
|
|
|
|
store := deviceview.NewStore()
|
|
service, err := deviceview.NewService(store, slog.Default(), 90*time.Second)
|
|
if err != nil {
|
|
return fmt.Errorf("create device-view service: %w", err)
|
|
}
|
|
|
|
telemetryFilter, err := contracts.DeviceFilter(contracts.KindTelemetry)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
responsesFilter, err := contracts.DeviceFilter(contracts.KindResponses)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
commandTopic, err := contracts.DeviceTopic(cfg.MQTT.DeviceID, contracts.KindCommands)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
inbox := make(chan mqtt.Message, 16)
|
|
manager, err := mqtt.Start(transportCtx, cfg, []contracts.TopicFilter{telemetryFilter, responsesFilter}, inbox)
|
|
if err != nil {
|
|
return fmt.Errorf("failed to start MQTT client: %w", err)
|
|
}
|
|
defer mqtt.Stop(manager)
|
|
|
|
telemetryPool := dispatch.New(context.Background(), 2, 16)
|
|
importantPool := dispatch.New(context.Background(), 2, 32)
|
|
receiver := &dispatch.Receiver{
|
|
Telemetry: telemetryPool,
|
|
Important: importantPool,
|
|
Process: service.HandleMessage,
|
|
}
|
|
defer func() {
|
|
drainCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
|
defer cancel()
|
|
started := time.Now()
|
|
if err := dispatch.Shutdown(drainCtx, telemetryPool, importantPool); err != nil {
|
|
log.Printf("dispatcher shutdown incomplete: %v", err)
|
|
}
|
|
log.Printf("dispatcher drain finished elapsed=%s telemetry=%+v important=%+v",
|
|
time.Since(started), telemetryPool.Stats(), importantPool.Stats())
|
|
}()
|
|
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
runID := time.Now().UnixNano()
|
|
sequence := 0
|
|
|
|
for {
|
|
select {
|
|
case <-lifecycleCtx.Done():
|
|
log.Print("shutting down; draining accepted work before MQTT disconnect")
|
|
return nil
|
|
case message := <-inbox:
|
|
outcome, submitErr := receiver.Submit(lifecycleCtx, message)
|
|
if submitErr != nil || outcome != dispatch.Accepted {
|
|
slog.Warn("message admission rejected", "topic", message.Topic, "outcome", outcome, "error", submitErr)
|
|
}
|
|
case <-ticker.C:
|
|
sequence++
|
|
id := fmt.Sprintf("ping-%d-%d", runID, sequence)
|
|
payload, marshalErr := json.Marshal(contracts.Command{
|
|
SchemaVersion: contracts.SchemaVersionOne,
|
|
MessageID: id,
|
|
DeviceID: cfg.MQTT.DeviceID,
|
|
OccurredAt: time.Now().UTC().Format(time.RFC3339),
|
|
CommandID: id,
|
|
Command: "ping",
|
|
})
|
|
if marshalErr != nil {
|
|
return fmt.Errorf("failed to marshal payload: %w", marshalErr)
|
|
}
|
|
if publishErr := mqtt.Publish(lifecycleCtx, manager, commandTopic, payload); publishErr != nil {
|
|
log.Printf("command publication failed or outcome uncertain: %v", publishErr)
|
|
}
|
|
}
|
|
}
|
|
}
|