shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(go): decouple mqtt message publishing from shutdown context

Browse Source
This commit is contained in:
lingniu
2026-07-02 13:14:15 +08:00
parent d10d5d75c3
commit a80eb2eab4
2 changed files with 45 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
go/vehicle-gateway/internal/gateway/mqtt_client.go
View File

@@ -176,6 +176,9 @@ func (c *MQTTClient) buildTLSConfig() (*tls.Config, error) {
} }
func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) { func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []byte) {
messageCtx, cancelMessage := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout)
defer cancelMessage()
receivedAtMS := time.Now().UnixMilli() receivedAtMS := time.Now().UnixMilli()
env, err := yutongmqtt.ParseMessage(c.cfg.EndpointName, topic, payload, receivedAtMS) env, err := yutongmqtt.ParseMessage(c.cfg.EndpointName, topic, payload, receivedAtMS)
if err != nil { if err != nil {
@@ -193,7 +196,7 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []
} }
env.EventID = env.StableEventID() env.EventID = env.StableEventID()
} else { } else {
resolved, resolveErr := c.cfg.Resolver.Resolve(ctx, env) resolved, resolveErr := c.cfg.Resolver.Resolve(messageCtx, env)
if resolveErr != nil { if resolveErr != nil {
c.cfg.Logger.Warn("mqtt identity resolve failed", "topic", topic, "event_id", env.StableEventID(), "error", resolveErr) c.cfg.Logger.Warn("mqtt identity resolve failed", "topic", topic, "event_id", env.StableEventID(), "error", resolveErr)
if env.Parsed == nil { if env.Parsed == nil {
@@ -205,14 +208,14 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload []
env = resolved env = resolved
} }
} }
if err := c.cfg.Sink.PublishRaw(ctx, env); err != nil { if err := c.cfg.Sink.PublishRaw(messageCtx, env); err != nil {
c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err) c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
return return
} }
if env.ParseStatus == envelope.ParseBadFrame { if env.ParseStatus == envelope.ParseBadFrame {
return return
} }
if err := c.cfg.Sink.PublishUnified(ctx, env); err != nil { if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil {
c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err) c.cfg.Logger.Error("publish mqtt unified failed", "topic", topic, "event_id", env.StableEventID(), "error", err)
} }
} }

39
go/vehicle-gateway/internal/gateway/mqtt_client_test.go
View File

@@ -69,6 +69,45 @@ func TestMQTTClientHandleBadPayloadPublishesOnlyRaw(t *testing.T) {
} }
} }
func TestMQTTClientUsesUncancelledMessageContextForReceivedMessage(t *testing.T) {
parent, cancel := context.WithCancel(context.Background())
cancel()
resolver := &contextCheckingResolver{}
sink := &contextCheckingSink{}
client, err := NewMQTTClient(MQTTClientConfig{
EndpointName: "endpoint-a",
Broker: "tcp://127.0.0.1:1883",
ClientID: "test-client",
Topics: []string{"/ytforward/shln/+"},
Sink: sink,
Resolver: resolver,
Logger: slog.New(slog.NewTextHandler(testWriter{t: t}, nil)),
})
if err != nil {
t.Fatalf("NewMQTTClient() error = %v", err)
}
client.handleMessage(parent, "/ytforward/shln/dev1", []byte(`{
"device":"LTEST000000000001",
"time":"20260413100000",
"data":{"METER_SPEED":52.3,"TOTAL_MILEAGE":123456.7}
}`))
if resolver.ctxErr != nil {
t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr)
}
if sink.rawCtxErr != nil {
t.Fatalf("raw publish saw cancelled context: %v", sink.rawCtxErr)
}
if sink.unifiedCtxErr != nil {
t.Fatalf("unified publish saw cancelled context: %v", sink.unifiedCtxErr)
}
if sink.rawCount != 1 || sink.unifiedCount != 1 {
t.Fatalf("raw=%d unified=%d", sink.rawCount, sink.unifiedCount)
}
}
func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) { func TestMQTTClientBuildOptionsLoadsTLSCertificates(t *testing.T) {
dir := t.TempDir() dir := t.TempDir()
caPath, certPath, keyPath := writeTestTLSMaterial(t, dir) caPath, certPath, keyPath := writeTestTLSMaterial(t, dir)