From 535acdb2a6d3038cf18a9eff057116fef3e5456e Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 14:08:32 +0800 Subject: [PATCH] feat(go): publish realtime fields stream --- go/vehicle-gateway/cmd/gateway/main.go | 12 ++++- .../cmd/nats-fast-writer/main.go | 3 ++ .../cmd/nats-kafka-bridge/main.go | 9 ++-- .../cmd/nats-kafka-bridge/main_test.go | 9 ++-- go/vehicle-gateway/cmd/stat-writer/main.go | 2 +- .../cmd/stat-writer/main_test.go | 2 +- .../internal/envelope/realtime.go | 51 +++++++++++++++++++ .../internal/eventbus/async_sink.go | 6 +++ .../internal/eventbus/async_sink_test.go | 9 ++++ .../internal/eventbus/durable_sink.go | 17 ++++++- .../internal/eventbus/durable_sink_test.go | 13 +++++ .../internal/eventbus/kafka_sink.go | 29 ++++++++++- .../internal/eventbus/log_sink.go | 5 ++ .../internal/eventbus/nats_sink.go | 21 ++++++++ .../internal/eventbus/retry_sink.go | 10 ++++ .../internal/eventbus/retry_sink_test.go | 4 ++ .../internal/gateway/mqtt_client.go | 9 ++++ .../internal/gateway/tcp_server.go | 9 ++++ .../internal/gateway/tcp_server_test.go | 14 +++++ go/vehicle-gateway/internal/realtime/kv.go | 45 ++++++++++++++++ .../internal/realtime/repository.go | 6 +++ .../internal/realtime/snapshot_writer.go | 3 ++ go/vehicle-gateway/internal/topics/topics.go | 11 ++-- 23 files changed, 284 insertions(+), 15 deletions(-) create mode 100644 go/vehicle-gateway/internal/envelope/realtime.go diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index 27f4d75c..6a5b1f91 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -186,7 +186,7 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) }) logger.Info("nats async publish enabled", "queue_size", queueSize, "workers", workers, "publish_timeout_ms", timeout.Milliseconds()) } - logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject, "publish_unified_enabled", envBool("PUBLISH_UNIFIED_ENABLED", false)) + logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "unified_subject", natsSinkConfigFromEnv().UnifiedSubject, "publish_unified_enabled", envBool("PUBLISH_UNIFIED_ENABLED", false), "publish_fields_enabled", true) return out, nil } brokers := splitCSV(os.Getenv("KAFKA_BROKERS")) @@ -201,6 +201,11 @@ func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) envelope.ProtocolJT808: env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808), envelope.ProtocolYutongMQTT: env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT), }, + FieldsTopics: map[envelope.Protocol]string{ + envelope.ProtocolGB32960: env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960), + envelope.ProtocolJT808: env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808), + envelope.ProtocolYutongMQTT: env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT), + }, UnifiedTopic: env("KAFKA_TOPIC_UNIFIED", topics.Unified), }) if err != nil { @@ -248,6 +253,11 @@ func natsSinkConfigFromEnv() eventbus.NATSConfig { envelope.ProtocolJT808: env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)), envelope.ProtocolYutongMQTT: env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)), }, + FieldsSubjects: map[envelope.Protocol]string{ + envelope.ProtocolGB32960: env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)), + envelope.ProtocolJT808: env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)), + envelope.ProtocolYutongMQTT: env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)), + }, UnifiedSubject: env("NATS_SUBJECT_UNIFIED", env("KAFKA_TOPIC_UNIFIED", topics.Unified)), } } diff --git a/go/vehicle-gateway/cmd/nats-fast-writer/main.go b/go/vehicle-gateway/cmd/nats-fast-writer/main.go index 56f6d41f..a37c1bae 100644 --- a/go/vehicle-gateway/cmd/nats-fast-writer/main.go +++ b/go/vehicle-gateway/cmd/nats-fast-writer/main.go @@ -153,6 +153,9 @@ func loadConfig() config { env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)), env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)), env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)), + env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)), + env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)), + env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)), }, ","))) return config{ NATSURL: env("NATS_URL", "nats://127.0.0.1:4222"), diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go index d13b6a46..2bcad936 100644 --- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main.go @@ -102,9 +102,12 @@ type config struct { func loadConfig() config { route := map[string]string{ - env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)): env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960), - env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)): env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808), - env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT), + env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)): env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960), + env("NATS_SUBJECT_JT808_RAW", env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808)): env("KAFKA_TOPIC_JT808_RAW", topics.RawJT808), + env("NATS_SUBJECT_YUTONG_MQTT_RAW", env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_RAW", topics.RawYutongMQTT), + env("NATS_SUBJECT_GB32960_FIELDS", env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960)): env("KAFKA_TOPIC_GB32960_FIELDS", topics.FieldsGB32960), + env("NATS_SUBJECT_JT808_FIELDS", env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808)): env("KAFKA_TOPIC_JT808_FIELDS", topics.FieldsJT808), + env("NATS_SUBJECT_YUTONG_MQTT_FIELDS", env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT)): env("KAFKA_TOPIC_YUTONG_MQTT_FIELDS", topics.FieldsYutongMQTT), } if unifiedSubject, unifiedTopic, ok := unifiedRouteFromEnv(); ok { route[unifiedSubject] = unifiedTopic diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go index cc0e2087..c8fb6773 100644 --- a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go @@ -117,9 +117,12 @@ func TestLoadConfigDefaultsToGoSubjectRoutes(t *testing.T) { cfg := loadConfig() want := map[string]string{ - "vehicle.raw.go.gb32960.v1": "vehicle.raw.go.gb32960.v1", - "vehicle.raw.go.jt808.v1": "vehicle.raw.go.jt808.v1", - "vehicle.raw.go.yutong-mqtt.v1": "vehicle.raw.go.yutong-mqtt.v1", + "vehicle.raw.go.gb32960.v1": "vehicle.raw.go.gb32960.v1", + "vehicle.raw.go.jt808.v1": "vehicle.raw.go.jt808.v1", + "vehicle.raw.go.yutong-mqtt.v1": "vehicle.raw.go.yutong-mqtt.v1", + "vehicle.fields.go.gb32960.v1": "vehicle.fields.go.gb32960.v1", + "vehicle.fields.go.jt808.v1": "vehicle.fields.go.jt808.v1", + "vehicle.fields.go.yutong-mqtt.v1": "vehicle.fields.go.yutong-mqtt.v1", } if len(cfg.Route) != len(want) { t.Fatalf("route len = %d, want %d: %#v", len(cfg.Route), len(want), cfg.Route) diff --git a/go/vehicle-gateway/cmd/stat-writer/main.go b/go/vehicle-gateway/cmd/stat-writer/main.go index cb5c7d18..b590a595 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main.go +++ b/go/vehicle-gateway/cmd/stat-writer/main.go @@ -143,7 +143,7 @@ func loadConfig() config { } return config{ KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")), - KafkaTopics: splitCSV(env("KAFKA_TOPICS", strings.Join([]string{topics.RawGB32960, topics.RawJT808, topics.RawYutongMQTT}, ","))), + KafkaTopics: splitCSV(env("KAFKA_TOPICS", strings.Join([]string{topics.FieldsGB32960, topics.FieldsJT808, topics.FieldsYutongMQTT}, ","))), KafkaGroup: env("KAFKA_GROUP", "go-stat-writer"), MySQLDSN: env("MYSQL_DSN", ""), EnsureSchema: env("MYSQL_ENSURE_SCHEMA", "true") != "false", diff --git a/go/vehicle-gateway/cmd/stat-writer/main_test.go b/go/vehicle-gateway/cmd/stat-writer/main_test.go index 945c14ba..fd18bd7d 100644 --- a/go/vehicle-gateway/cmd/stat-writer/main_test.go +++ b/go/vehicle-gateway/cmd/stat-writer/main_test.go @@ -78,7 +78,7 @@ func TestLoadConfigDefaultsToGoRawTopics(t *testing.T) { cfg := loadConfig() got := strings.Join(cfg.KafkaTopics, ",") - want := "vehicle.raw.go.gb32960.v1,vehicle.raw.go.jt808.v1,vehicle.raw.go.yutong-mqtt.v1" + want := "vehicle.fields.go.gb32960.v1,vehicle.fields.go.jt808.v1,vehicle.fields.go.yutong-mqtt.v1" if got != want { t.Fatalf("KafkaTopics = %q, want %q", got, want) } diff --git a/go/vehicle-gateway/internal/envelope/realtime.go b/go/vehicle-gateway/internal/envelope/realtime.go new file mode 100644 index 00000000..04ca875e --- /dev/null +++ b/go/vehicle-gateway/internal/envelope/realtime.go @@ -0,0 +1,51 @@ +package envelope + +import "strings" + +func IsRealtimeTelemetryFrame(env FrameEnvelope) bool { + if env.ParseStatus == ParseBadFrame { + return false + } + switch env.Protocol { + case ProtocolGB32960: + command := strings.TrimSpace(env.MessageID) + if command == "" && env.Parsed != nil { + if header, ok := env.Parsed["header"].(map[string]any); ok { + command = strings.TrimSpace(stringAny(header["command"])) + } + } + if strings.EqualFold(command, "0x02") || strings.EqualFold(command, "0x03") { + return true + } + _, hasDataUnits := env.Parsed["data_units"] + return hasDataUnits || hasRealtimeField(env) + case ProtocolJT808: + if strings.EqualFold(strings.TrimSpace(env.MessageID), "0x0200") { + return true + } + if _, ok := env.Parsed["location"]; ok { + return true + } + return hasRealtimeField(env) + case ProtocolYutongMQTT: + data, ok := env.Parsed["data"].(map[string]any) + return ok && len(data) > 0 + default: + return false + } +} + +func hasRealtimeField(env FrameEnvelope) bool { + return env.Fields[FieldLatitude] != nil || + env.Fields[FieldLongitude] != nil || + env.Fields[FieldTotalMileageKM] != nil || + env.Fields[FieldSpeedKMH] != nil || + env.Fields[FieldSOCPercent] != nil +} + +func stringAny(value any) string { + if text, ok := value.(string); ok { + return text + } + return "" +} diff --git a/go/vehicle-gateway/internal/eventbus/async_sink.go b/go/vehicle-gateway/internal/eventbus/async_sink.go index 5b6d2591..8315427f 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink.go @@ -75,6 +75,10 @@ func (s *AsyncSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelo return s.enqueue(ctx, asyncJob{kind: "unified", env: env}) } +func (s *AsyncSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error { + return s.enqueue(ctx, asyncJob{kind: "fields", env: env}) +} + func (s *AsyncSink) Close() error { s.closeOnce.Do(func() { close(s.closed) @@ -110,6 +114,8 @@ func (s *AsyncSink) worker() { err = s.delegate.PublishRaw(ctx, job.env) case "unified": err = s.delegate.PublishUnified(ctx, job.env) + case "fields": + err = s.delegate.PublishFields(ctx, job.env) } cancel() if err != nil && s.onError != nil { diff --git a/go/vehicle-gateway/internal/eventbus/async_sink_test.go b/go/vehicle-gateway/internal/eventbus/async_sink_test.go index fc58548c..4264806c 100644 --- a/go/vehicle-gateway/internal/eventbus/async_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/async_sink_test.go @@ -73,6 +73,10 @@ func (s *blockingSink) PublishUnified(context.Context, envelope.FrameEnvelope) e return nil } +func (s *blockingSink) PublishFields(context.Context, envelope.FrameEnvelope) error { + return nil +} + func (s *blockingSink) Close() error { s.release() return nil @@ -100,6 +104,11 @@ func (s *orderedSink) PublishUnified(context.Context, envelope.FrameEnvelope) er return nil } +func (s *orderedSink) PublishFields(context.Context, envelope.FrameEnvelope) error { + s.calls = append(s.calls, "fields") + return nil +} + func (s *orderedSink) Close() error { return nil } diff --git a/go/vehicle-gateway/internal/eventbus/durable_sink.go b/go/vehicle-gateway/internal/eventbus/durable_sink.go index 1a2b9f3f..8dd193c9 100644 --- a/go/vehicle-gateway/internal/eventbus/durable_sink.go +++ b/go/vehicle-gateway/internal/eventbus/durable_sink.go @@ -79,6 +79,16 @@ func (s *DurableSink) PublishUnified(ctx context.Context, env envelope.FrameEnve return s.spool("unified", env) } +func (s *DurableSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error { + if s.isRawPending(env) { + return s.spool("fields", env) + } + if err := s.delegate.PublishFields(ctx, env); err == nil { + return nil + } + return s.spool("fields", env) +} + func (s *DurableSink) ReplayOnce(ctx context.Context) error { return s.replay(ctx, 0) } @@ -147,7 +157,10 @@ func durableKindOrder(kind string) int { if kind == "unified" { return 1 } - return 2 + if kind == "fields" { + return 2 + } + return 3 } func (s *DurableSink) ReplayLoop(ctx context.Context, interval time.Duration, onError func(error)) { @@ -184,6 +197,8 @@ func (s *DurableSink) publishRecord(ctx context.Context, record durableRecord) e return s.delegate.PublishRaw(ctx, record.Envelope) case "unified": return s.delegate.PublishUnified(ctx, record.Envelope) + case "fields": + return s.delegate.PublishFields(ctx, record.Envelope) default: return errUnknownRecordKind(record.Kind) } diff --git a/go/vehicle-gateway/internal/eventbus/durable_sink_test.go b/go/vehicle-gateway/internal/eventbus/durable_sink_test.go index eaddbe9a..902600a3 100644 --- a/go/vehicle-gateway/internal/eventbus/durable_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/durable_sink_test.go @@ -300,6 +300,11 @@ func (s *scriptedSink) PublishUnified(context.Context, envelope.FrameEnvelope) e return nil } +func (s *scriptedSink) PublishFields(context.Context, envelope.FrameEnvelope) error { + s.calls = append(s.calls, "fields") + return nil +} + func (s *scriptedSink) Close() error { return nil } @@ -323,6 +328,10 @@ func (s *contextCheckingReplaySink) PublishUnified(ctx context.Context, _ envelo return s.unifiedCtxErr } +func (s *contextCheckingReplaySink) PublishFields(ctx context.Context, _ envelope.FrameEnvelope) error { + return ctx.Err() +} + func (s *contextCheckingReplaySink) Close() error { return nil } @@ -350,6 +359,10 @@ func (s *batchRecordingSink) PublishUnified(context.Context, envelope.FrameEnvel return nil } +func (s *batchRecordingSink) PublishFields(context.Context, envelope.FrameEnvelope) error { + return nil +} + func (s *batchRecordingSink) Close() error { return nil } diff --git a/go/vehicle-gateway/internal/eventbus/kafka_sink.go b/go/vehicle-gateway/internal/eventbus/kafka_sink.go index 4f32f4d2..a81a34d3 100644 --- a/go/vehicle-gateway/internal/eventbus/kafka_sink.go +++ b/go/vehicle-gateway/internal/eventbus/kafka_sink.go @@ -14,12 +14,14 @@ import ( type Sink interface { PublishRaw(context.Context, envelope.FrameEnvelope) error PublishUnified(context.Context, envelope.FrameEnvelope) error + PublishFields(context.Context, envelope.FrameEnvelope) error Close() error } type KafkaSink struct { writer kafkaWriter rawTopics map[envelope.Protocol]string + fieldsTopics map[envelope.Protocol]string unifiedTopic string } @@ -31,6 +33,7 @@ type kafkaWriter interface { type KafkaConfig struct { Brokers []string RawTopics map[envelope.Protocol]string + FieldsTopics map[envelope.Protocol]string UnifiedTopic string } @@ -58,11 +61,21 @@ func newKafkaSinkWithWriter(writer kafkaWriter, cfg KafkaConfig) *KafkaSink { rawTopics[protocol] = topic } } + fieldsTopics := map[envelope.Protocol]string{ + envelope.ProtocolGB32960: topics.FieldsGB32960, + envelope.ProtocolJT808: topics.FieldsJT808, + envelope.ProtocolYutongMQTT: topics.FieldsYutongMQTT, + } + for protocol, topic := range cfg.FieldsTopics { + if topic != "" { + fieldsTopics[protocol] = topic + } + } unifiedTopic := cfg.UnifiedTopic if unifiedTopic == "" { unifiedTopic = topics.Unified } - return &KafkaSink{writer: writer, rawTopics: rawTopics, unifiedTopic: unifiedTopic} + return &KafkaSink{writer: writer, rawTopics: rawTopics, fieldsTopics: fieldsTopics, unifiedTopic: unifiedTopic} } func (s *KafkaSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error { @@ -77,6 +90,14 @@ func (s *KafkaSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelo return s.publish(ctx, s.unifiedTopic, env) } +func (s *KafkaSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error { + topic, ok := s.fieldsTopics[env.Protocol] + if !ok || topic == "" { + return fmt.Errorf("fields topic not configured for protocol %s", env.Protocol) + } + return s.publish(ctx, topic, env) +} + func (s *KafkaSink) PublishRecords(ctx context.Context, records []durableRecord) error { messages := make([]kafka.Message, 0, len(records)) for _, record := range records { @@ -121,6 +142,12 @@ func (s *KafkaSink) topicForRecord(record durableRecord) (string, error) { return topic, nil case "unified": return s.unifiedTopic, nil + case "fields": + topic, ok := s.fieldsTopics[record.Envelope.Protocol] + if !ok || topic == "" { + return "", fmt.Errorf("fields topic not configured for protocol %s", record.Envelope.Protocol) + } + return topic, nil default: return "", fmt.Errorf("unknown durable record kind %q", record.Kind) } diff --git a/go/vehicle-gateway/internal/eventbus/log_sink.go b/go/vehicle-gateway/internal/eventbus/log_sink.go index 68eed2e5..bf930501 100644 --- a/go/vehicle-gateway/internal/eventbus/log_sink.go +++ b/go/vehicle-gateway/internal/eventbus/log_sink.go @@ -28,6 +28,11 @@ func (s *LogSink) PublishUnified(_ context.Context, env envelope.FrameEnvelope) return nil } +func (s *LogSink) PublishFields(_ context.Context, env envelope.FrameEnvelope) error { + s.logger.Info("fields envelope", "protocol", env.Protocol, "event_id", env.StableEventID(), "vehicle_key", env.VehicleKey(), "message_id", env.MessageID, "field_count", len(env.Fields)) + return nil +} + func (s *LogSink) Close() error { return nil } diff --git a/go/vehicle-gateway/internal/eventbus/nats_sink.go b/go/vehicle-gateway/internal/eventbus/nats_sink.go index c0629dfa..46d216a2 100644 --- a/go/vehicle-gateway/internal/eventbus/nats_sink.go +++ b/go/vehicle-gateway/internal/eventbus/nats_sink.go @@ -16,6 +16,7 @@ type NATSConfig struct { URL string Name string RawSubjects map[envelope.Protocol]string + FieldsSubjects map[envelope.Protocol]string UnifiedSubject string } @@ -23,6 +24,7 @@ type NATSSink struct { conn *nats.Conn publisher natsPublisher rawSubjects map[envelope.Protocol]string + fieldsSubjects map[envelope.Protocol]string unifiedSubject string } @@ -65,6 +67,16 @@ func newNATSSinkWithPublisher(publisher natsPublisher, cfg NATSConfig) *NATSSink rawSubjects[protocol] = subject } } + fieldsSubjects := map[envelope.Protocol]string{ + envelope.ProtocolGB32960: topics.FieldsGB32960, + envelope.ProtocolJT808: topics.FieldsJT808, + envelope.ProtocolYutongMQTT: topics.FieldsYutongMQTT, + } + for protocol, subject := range cfg.FieldsSubjects { + if subject != "" { + fieldsSubjects[protocol] = subject + } + } unifiedSubject := cfg.UnifiedSubject if unifiedSubject == "" { unifiedSubject = topics.Unified @@ -72,6 +84,7 @@ func newNATSSinkWithPublisher(publisher natsPublisher, cfg NATSConfig) *NATSSink return &NATSSink{ publisher: publisher, rawSubjects: rawSubjects, + fieldsSubjects: fieldsSubjects, unifiedSubject: unifiedSubject, } } @@ -91,6 +104,14 @@ func (s *NATSSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelop return s.publish(ctx, s.unifiedSubject, env) } +func (s *NATSSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error { + subject, ok := s.fieldsSubjects[env.Protocol] + if !ok || subject == "" { + return fmt.Errorf("fields subject not configured for protocol %s", env.Protocol) + } + return s.publish(ctx, subject, env) +} + func (s *NATSSink) Close() error { if s == nil || s.conn == nil { return nil diff --git a/go/vehicle-gateway/internal/eventbus/retry_sink.go b/go/vehicle-gateway/internal/eventbus/retry_sink.go index 4e2ed05d..f105cc97 100644 --- a/go/vehicle-gateway/internal/eventbus/retry_sink.go +++ b/go/vehicle-gateway/internal/eventbus/retry_sink.go @@ -40,6 +40,12 @@ func (s *RetryingSink) PublishUnified(ctx context.Context, env envelope.FrameEnv }) } +func (s *RetryingSink) PublishFields(ctx context.Context, env envelope.FrameEnvelope) error { + return s.withRetry(ctx, func(attemptCtx context.Context) error { + return s.delegate.PublishFields(attemptCtx, env) + }) +} + func (s *RetryingSink) PublishRecords(ctx context.Context, records []durableRecord) error { if publisher, ok := s.delegate.(recordPublishingSink); ok { return s.withRetry(ctx, func(attemptCtx context.Context) error { @@ -56,6 +62,10 @@ func (s *RetryingSink) PublishRecords(ctx context.Context, records []durableReco if err := s.PublishUnified(ctx, record.Envelope); err != nil { return err } + case "fields": + if err := s.PublishFields(ctx, record.Envelope); err != nil { + return err + } default: return errUnknownRecordKind(record.Kind) } diff --git a/go/vehicle-gateway/internal/eventbus/retry_sink_test.go b/go/vehicle-gateway/internal/eventbus/retry_sink_test.go index ca5641ec..dc6f92b4 100644 --- a/go/vehicle-gateway/internal/eventbus/retry_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/retry_sink_test.go @@ -75,6 +75,10 @@ func (s *flakySink) PublishUnified(context.Context, envelope.FrameEnvelope) erro return nil } +func (s *flakySink) PublishFields(context.Context, envelope.FrameEnvelope) error { + return nil +} + func (s *flakySink) Close() error { return nil } diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index ed7873b0..355dea7c 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -18,6 +18,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/yutongmqtt" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" ) type MQTTClientConfig struct { @@ -219,6 +220,14 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] if env.ParseStatus == envelope.ParseBadFrame { return } + if fieldsEnv, ok := realtime.BuildFieldsEnvelope(env); ok { + if err := c.cfg.Sink.PublishFields(messageCtx, fieldsEnv); err != nil { + c.recordPublishMetric("fields", "error") + c.cfg.Logger.Error("publish mqtt fields failed", "topic", topic, "event_id", env.StableEventID(), "error", err) + return + } + c.recordPublishMetric("fields", "ok") + } if c.cfg.PublishUnified { if err := c.cfg.Sink.PublishUnified(messageCtx, env); err != nil { c.recordPublishMetric("unified", "error") diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 5b4bd78f..c3f6cd7a 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -16,6 +16,7 @@ import ( "lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime" ) type FrameExtractor func([]byte) (frames [][]byte, remainder []byte, err error) @@ -235,6 +236,14 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, if env.ParseStatus == envelope.ParseBadFrame { return } + if fieldsEnv, ok := realtime.BuildFieldsEnvelope(env); ok { + if err := s.sink.PublishFields(frameCtx, fieldsEnv); err != nil { + s.recordPublishMetric("fields", "error") + s.logger.Error("publish fields failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err) + return + } + s.recordPublishMetric("fields", "ok") + } if s.publishUnified { if err := s.sink.PublishUnified(frameCtx, env); err != nil { s.recordPublishMetric("unified", "error") diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index a5ee19af..d7fcf6d9 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -281,8 +281,10 @@ func (r *contextCheckingResolver) Resolve(ctx context.Context, env envelope.Fram type contextCheckingSink struct { rawCtxErr error unifiedCtxErr error + fieldsCtxErr error rawCount int unifiedCount int + fieldsCount int } func (s *contextCheckingSink) PublishRaw(ctx context.Context, _ envelope.FrameEnvelope) error { @@ -297,6 +299,12 @@ func (s *contextCheckingSink) PublishUnified(ctx context.Context, _ envelope.Fra return s.unifiedCtxErr } +func (s *contextCheckingSink) PublishFields(ctx context.Context, _ envelope.FrameEnvelope) error { + s.fieldsCtxErr = ctx.Err() + s.fieldsCount++ + return s.fieldsCtxErr +} + func (s *contextCheckingSink) Close() error { return nil } @@ -331,6 +339,7 @@ func runPipe(t *testing.T, server *TCPServer) (net.Conn, <-chan struct{}) { type recordingSink struct { raw []envelope.FrameEnvelope unified []envelope.FrameEnvelope + fields []envelope.FrameEnvelope } func (s *recordingSink) PublishRaw(_ context.Context, env envelope.FrameEnvelope) error { @@ -343,6 +352,11 @@ func (s *recordingSink) PublishUnified(_ context.Context, env envelope.FrameEnve return nil } +func (s *recordingSink) PublishFields(_ context.Context, env envelope.FrameEnvelope) error { + s.fields = append(s.fields, env) + return nil +} + func (s *recordingSink) Close() error { return nil } diff --git a/go/vehicle-gateway/internal/realtime/kv.go b/go/vehicle-gateway/internal/realtime/kv.go index e35a3af3..67d427fe 100644 --- a/go/vehicle-gateway/internal/realtime/kv.go +++ b/go/vehicle-gateway/internal/realtime/kv.go @@ -22,6 +22,51 @@ type RealtimeKVField struct { EventID string } +func BuildFieldsEnvelope(env envelope.FrameEnvelope) (envelope.FrameEnvelope, bool) { + if !envelope.IsRealtimeTelemetryFrame(env) { + return envelope.FrameEnvelope{}, false + } + rows := realtimeKVFields(env, env.Parsed) + fields := make(map[string]any, len(rows)+len(env.Fields)) + for _, row := range rows { + key := row.Domain + if strings.TrimSpace(row.Field) != "" { + key += "." + row.Field + } + fields[key] = row.Value + } + for key, value := range env.Fields { + if strings.TrimSpace(key) != "" { + fields[key] = value + } + } + if len(fields) == 0 { + return envelope.FrameEnvelope{}, false + } + out := envelope.FrameEnvelope{ + EventID: env.StableEventID() + ":fields", + TraceID: env.TraceID, + Protocol: env.Protocol, + MessageID: env.MessageID, + Sequence: env.Sequence, + VIN: env.VIN, + VehicleKeyHint: env.VehicleKeyHint, + Phone: env.Phone, + DeviceID: env.DeviceID, + Plate: env.Plate, + SourceEndpoint: env.SourceEndpoint, + EventTimeMS: env.EventTimeMS, + ReceivedAtMS: env.ReceivedAtMS, + Fields: fields, + Parsed: map[string]any{ + "source_event_id": env.StableEventID(), + "field_mapping": realtimeFieldMappingVersion, + }, + ParseStatus: envelope.ParseOK, + } + return out, true +} + type protocolFieldMapping struct { Prefix string GBDataUnits bool diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index f15ba6cb..52c6b8e1 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -28,6 +28,9 @@ func NewRepository(client *redis.Client, cfg Config) *Repository { } func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) error { + if !envelope.IsRealtimeTelemetryFrame(env) { + return nil + } vin := strings.TrimSpace(env.VIN) if vin == "" { return nil @@ -40,6 +43,9 @@ func (r *Repository) FastUpdate(ctx context.Context, env envelope.FrameEnvelope) } func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) error { + if !envelope.IsRealtimeTelemetryFrame(env) { + return nil + } vin := strings.TrimSpace(env.VIN) if vin == "" { return nil diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 682eedfe..57d8c411 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -118,6 +118,9 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { } func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error { + if !envelope.IsRealtimeTelemetryFrame(env) { + return nil + } vin := strings.TrimSpace(env.VIN) if vin == "" { return nil diff --git a/go/vehicle-gateway/internal/topics/topics.go b/go/vehicle-gateway/internal/topics/topics.go index 8d42a7fd..cf3297d8 100644 --- a/go/vehicle-gateway/internal/topics/topics.go +++ b/go/vehicle-gateway/internal/topics/topics.go @@ -1,8 +1,11 @@ package topics const ( - RawGB32960 = "vehicle.raw.go.gb32960.v1" - RawJT808 = "vehicle.raw.go.jt808.v1" - RawYutongMQTT = "vehicle.raw.go.yutong-mqtt.v1" - Unified = "vehicle.event.go.unified.v1" + RawGB32960 = "vehicle.raw.go.gb32960.v1" + RawJT808 = "vehicle.raw.go.jt808.v1" + RawYutongMQTT = "vehicle.raw.go.yutong-mqtt.v1" + FieldsGB32960 = "vehicle.fields.go.gb32960.v1" + FieldsJT808 = "vehicle.fields.go.jt808.v1" + FieldsYutongMQTT = "vehicle.fields.go.yutong-mqtt.v1" + Unified = "vehicle.event.go.unified.v1" )