From c27150cc15acad7a16f58a0e019fc92980956a8f Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 15:07:44 +0800 Subject: [PATCH] feat(go): add nats kafka ingest bridge --- deploy/nats/docker-compose.yml | 12 + deploy/nats/nats-server.conf | 13 + .../2026-07-02-nats-kafka-ingest-refactor.md | 489 ++++++++++++++++++ go/vehicle-gateway/cmd/gateway/main.go | 40 ++ go/vehicle-gateway/cmd/gateway/main_test.go | 33 ++ .../cmd/nats-kafka-bridge/main.go | 260 ++++++++++ .../cmd/nats-kafka-bridge/main_test.go | 75 +++ go/vehicle-gateway/go.mod | 13 +- go/vehicle-gateway/go.sum | 26 +- .../internal/eventbus/nats_sink.go | 117 +++++ .../internal/eventbus/nats_sink_test.go | 55 ++ 11 files changed, 1121 insertions(+), 12 deletions(-) create mode 100644 deploy/nats/docker-compose.yml create mode 100644 deploy/nats/nats-server.conf create mode 100644 docs/superpowers/plans/2026-07-02-nats-kafka-ingest-refactor.md create mode 100644 go/vehicle-gateway/cmd/gateway/main_test.go create mode 100644 go/vehicle-gateway/cmd/nats-kafka-bridge/main.go create mode 100644 go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go create mode 100644 go/vehicle-gateway/internal/eventbus/nats_sink.go create mode 100644 go/vehicle-gateway/internal/eventbus/nats_sink_test.go diff --git a/deploy/nats/docker-compose.yml b/deploy/nats/docker-compose.yml new file mode 100644 index 00000000..a17ae174 --- /dev/null +++ b/deploy/nats/docker-compose.yml @@ -0,0 +1,12 @@ +services: + nats: + image: nats:2.10-alpine + container_name: lingniu-nats + restart: always + command: ["-c", "/etc/nats/nats-server.conf"] + ports: + - "172.17.111.56:4222:4222" + - "172.17.111.56:8222:8222" + volumes: + - /opt/lingniu-nats/conf/nats-server.conf:/etc/nats/nats-server.conf:ro + - /opt/lingniu-nats/data:/data diff --git a/deploy/nats/nats-server.conf b/deploy/nats/nats-server.conf new file mode 100644 index 00000000..6c665f2e --- /dev/null +++ b/deploy/nats/nats-server.conf @@ -0,0 +1,13 @@ +server_name: lingniu-vehicle-nats +port: 4222 +http_port: 8222 + +jetstream { + store_dir: "/data/jetstream" + max_mem_store: 1G + max_file_store: 200G +} + +authorization { + timeout: 2 +} diff --git a/docs/superpowers/plans/2026-07-02-nats-kafka-ingest-refactor.md b/docs/superpowers/plans/2026-07-02-nats-kafka-ingest-refactor.md new file mode 100644 index 00000000..11f418f7 --- /dev/null +++ b/docs/superpowers/plans/2026-07-02-nats-kafka-ingest-refactor.md @@ -0,0 +1,489 @@ +# NATS Kafka Ingest Refactor Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Move vehicle ingress publishing from direct Kafka writes to NATS JetStream, then bridge NATS to existing Kafka topics so downstream history, realtime, and stats services keep working. + +**Architecture:** `gateway` publishes parsed RAW and UNIFIED envelopes to NATS JetStream subjects. A new `nats-kafka-bridge` durable pull consumer reads from JetStream, writes batches to Kafka, and only ACKs NATS after Kafka succeeds. Existing Kafka consumers remain unchanged in this phase. + +**Tech Stack:** Go 1.26, `github.com/nats-io/nats.go`, NATS JetStream, Kafka via `segmentio/kafka-go`, existing `eventbus.Sink`, systemd single-binary deployment on ECS. + +--- + +## File Structure + +- Create `go/vehicle-gateway/internal/eventbus/nats_sink.go`: NATS JetStream implementation of `eventbus.Sink`. +- Create `go/vehicle-gateway/internal/eventbus/nats_sink_test.go`: topic/subject routing and envelope serialization tests. +- Modify `go/vehicle-gateway/cmd/gateway/main.go`: choose NATS sink when `NATS_URL` is configured; keep Kafka path as fallback. +- Create `go/vehicle-gateway/cmd/nats-kafka-bridge/main.go`: bridge process that creates JetStream stream/consumer, pulls NATS messages, writes Kafka batches, then ACKs. +- Create `go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go`: bridge batch ACK behavior and topic routing tests. +- Modify `go/vehicle-gateway/go.mod`: add `github.com/nats-io/nats.go`. +- Create `deploy/nats/nats-server.conf`: single-node JetStream config for Kafka ECS. +- Modify ECS deploy process: include `nats-kafka-bridge` binary and systemd service. + +## Task 1: Add NATS JetStream Sink + +**Files:** +- Create: `go/vehicle-gateway/internal/eventbus/nats_sink.go` +- Create: `go/vehicle-gateway/internal/eventbus/nats_sink_test.go` +- Modify: `go/vehicle-gateway/go.mod` + +- [x] **Step 1: Write failing subject routing test** + +Create `go/vehicle-gateway/internal/eventbus/nats_sink_test.go` with a fake JetStream publisher: + +```go +package eventbus + +import ( + "context" + "encoding/json" + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestNATSSinkRoutesRawAndUnifiedSubjects(t *testing.T) { + publisher := &recordingNATSPublisher{} + sink := newNATSSinkWithPublisher(publisher, NATSConfig{ + RawSubjects: map[envelope.Protocol]string{ + envelope.ProtocolJT808: "vehicle.raw.jt808.v1", + }, + UnifiedSubject: "vehicle.event.unified.v1", + }) + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"} + + if err := sink.PublishRaw(context.Background(), env); err != nil { + t.Fatalf("PublishRaw() error = %v", err) + } + if err := sink.PublishUnified(context.Background(), env); err != nil { + t.Fatalf("PublishUnified() error = %v", err) + } + + if got, want := publisher.messages[0].subject, "vehicle.raw.jt808.v1"; got != want { + t.Fatalf("raw subject = %q, want %q", got, want) + } + if got, want := publisher.messages[1].subject, "vehicle.event.unified.v1"; got != want { + t.Fatalf("unified subject = %q, want %q", got, want) + } + var decoded envelope.FrameEnvelope + if err := json.Unmarshal(publisher.messages[0].data, &decoded); err != nil { + t.Fatalf("raw payload is not envelope json: %v", err) + } + if decoded.Phone != "13307795425" { + t.Fatalf("decoded phone = %q", decoded.Phone) + } +} + +type recordingNATSPublisher struct { + messages []recordedNATSMessage +} + +type recordedNATSMessage struct { + subject string + data []byte +} + +func (p *recordingNATSPublisher) Publish(_ context.Context, subject string, data []byte, _ ...NATSPublishOption) error { + p.messages = append(p.messages, recordedNATSMessage{subject: subject, data: append([]byte(nil), data...)}) + return nil +} +``` + +Run: `go test ./internal/eventbus -run TestNATSSinkRoutesRawAndUnifiedSubjects -count=1` + +Expected: FAIL because `NATSConfig`, `newNATSSinkWithPublisher`, and `NATSPublishOption` do not exist. + +- [x] **Step 2: Implement NATS sink** + +Create `go/vehicle-gateway/internal/eventbus/nats_sink.go`: + +```go +package eventbus + +import ( + "context" + "errors" + "fmt" + + "github.com/nats-io/nats.go" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type NATSConfig struct { + URL string + Stream string + RawSubjects map[envelope.Protocol]string + UnifiedSubject string +} + +type NATSSink struct { + conn *nats.Conn + publisher natsPublisher + rawSubjects map[envelope.Protocol]string + unifiedSubject string +} + +type NATSPublishOption = nats.PubOpt + +type natsPublisher interface { + Publish(context.Context, string, []byte, ...NATSPublishOption) error +} + +func NewNATSSink(cfg NATSConfig) (*NATSSink, error) { + if cfg.URL == "" { + return nil, errors.New("nats url is required") + } + conn, err := nats.Connect(cfg.URL, nats.Name("lingniu-vehicle-gateway")) + if err != nil { + return nil, err + } + js, err := conn.JetStream() + if err != nil { + conn.Close() + return nil, err + } + sink := newNATSSinkWithPublisher(natsJetStreamPublisher{js: js}, cfg) + sink.conn = conn + return sink, nil +} + +func newNATSSinkWithPublisher(publisher natsPublisher, cfg NATSConfig) *NATSSink { + rawSubjects := map[envelope.Protocol]string{ + envelope.ProtocolGB32960: "vehicle.raw.gb32960.v1", + envelope.ProtocolJT808: "vehicle.raw.jt808.v1", + envelope.ProtocolYutongMQTT: "vehicle.raw.yutong-mqtt.v1", + } + for protocol, subject := range cfg.RawSubjects { + if subject != "" { + rawSubjects[protocol] = subject + } + } + unifiedSubject := cfg.UnifiedSubject + if unifiedSubject == "" { + unifiedSubject = "vehicle.event.unified.v1" + } + return &NATSSink{publisher: publisher, rawSubjects: rawSubjects, unifiedSubject: unifiedSubject} +} + +func (s *NATSSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error { + subject, ok := s.rawSubjects[env.Protocol] + if !ok || subject == "" { + return fmt.Errorf("raw subject not configured for protocol %s", env.Protocol) + } + return s.publish(ctx, subject, env) +} + +func (s *NATSSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error { + return s.publish(ctx, s.unifiedSubject, env) +} + +func (s *NATSSink) Close() error { + if s != nil && s.conn != nil { + s.conn.Drain() + s.conn.Close() + } + return nil +} + +func (s *NATSSink) publish(ctx context.Context, subject string, env envelope.FrameEnvelope) error { + payload, err := env.MarshalJSONBytes() + if err != nil { + return err + } + return s.publisher.Publish(ctx, subject, payload, nats.MsgId(env.StableEventID())) +} + +type natsJetStreamPublisher struct { + js nats.JetStreamContext +} + +func (p natsJetStreamPublisher) Publish(ctx context.Context, subject string, data []byte, opts ...NATSPublishOption) error { + _, err := p.js.Publish(subject, data, append(opts, nats.Context(ctx))...) + return err +} +``` + +- [x] **Step 3: Run eventbus tests** + +Run: `go get github.com/nats-io/nats.go@latest && go test ./internal/eventbus -count=1` + +Expected: PASS. + +## Task 2: Gateway Sink Selection + +**Files:** +- Modify: `go/vehicle-gateway/cmd/gateway/main.go` + +- [x] **Step 1: Write failing buildSink test** + +Create `go/vehicle-gateway/cmd/gateway/main_test.go` if it does not exist, or add: + +```go +func TestBuildSinkPrefersNATSWhenConfigured(t *testing.T) { + t.Setenv("NATS_URL", "nats://127.0.0.1:4222") + t.Setenv("KAFKA_BROKERS", "127.0.0.1:9092") + if got := chooseSinkMode(); got != "nats" { + t.Fatalf("sink mode = %q, want nats", got) + } +} +``` + +Run: `go test ./cmd/gateway -run TestBuildSinkPrefersNATSWhenConfigured -count=1` + +Expected: FAIL because `chooseSinkMode` does not exist. + +- [x] **Step 2: Implement mode choice and NATS sink branch** + +In `cmd/gateway/main.go`, add: + +```go +func chooseSinkMode() string { + if strings.TrimSpace(os.Getenv("NATS_URL")) != "" { + return "nats" + } + return "kafka" +} +``` + +At the top of `buildSink`, add: + +```go +if chooseSinkMode() == "nats" { + sink, err := eventbus.NewNATSSink(eventbus.NATSConfig{ + URL: env("NATS_URL", ""), + Stream: env("NATS_STREAM", "vehicle-ingest"), + RawSubjects: map[envelope.Protocol]string{ + envelope.ProtocolGB32960: env("NATS_SUBJECT_GB32960_RAW", "vehicle.raw.gb32960.v1"), + envelope.ProtocolJT808: env("NATS_SUBJECT_JT808_RAW", "vehicle.raw.jt808.v1"), + envelope.ProtocolYutongMQTT: env("NATS_SUBJECT_YUTONG_MQTT_RAW", "vehicle.raw.yutong-mqtt.v1"), + }, + UnifiedSubject: env("NATS_SUBJECT_UNIFIED", "vehicle.event.unified.v1"), + }) + if err != nil { + return nil, err + } + logger.Info("nats jetstream sink enabled", "url", env("NATS_URL", ""), "stream", env("NATS_STREAM", "vehicle-ingest")) + return eventbus.NewAsyncSink(sink, eventbus.AsyncConfig{ + QueueSize: envInt("NATS_ASYNC_QUEUE_SIZE", 100000), + Workers: envInt("NATS_ASYNC_WORKERS", 8), + OperationTimeout: time.Duration(envInt("NATS_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond, + OnError: func(err error) { + logger.Warn("nats async publish failed", "error", err) + }, + }), nil +} +``` + +- [x] **Step 3: Run gateway tests** + +Run: `go test ./cmd/gateway -count=1` + +Expected: PASS. + +## Task 3: NATS to Kafka Bridge + +**Files:** +- Create: `go/vehicle-gateway/cmd/nats-kafka-bridge/main.go` +- Create: `go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go` + +- [x] **Step 1: Write failing bridge test** + +Create a test with fake NATS messages and fake Kafka writer. It must prove that the bridge ACKs only after Kafka write succeeds: + +```go +func TestBridgeAcksAfterKafkaWrite(t *testing.T) { + nats := &fakeNATSBatchReader{messages: []bridgeMessage{ + {subject: "vehicle.raw.jt808.v1", data: []byte(`{"protocol":"JT808","phone":"13307795425"}`)}, + }} + kafka := &fakeKafkaBatchWriter{} + bridge := bridgeProcessor{nats: nats, kafka: kafka, topics: defaultBridgeTopics()} + + if err := bridge.processBatch(context.Background(), 10); err != nil { + t.Fatalf("processBatch() error = %v", err) + } + if kafka.writeCalls != 1 { + t.Fatalf("kafka write calls = %d, want 1", kafka.writeCalls) + } + if !nats.messages[0].acked { + t.Fatal("nats message was not acked") + } +} +``` + +Run: `go test ./cmd/nats-kafka-bridge -run TestBridgeAcksAfterKafkaWrite -count=1` + +Expected: FAIL because bridge types do not exist. + +- [x] **Step 2: Implement minimal bridge processor** + +Implement small interfaces in `main.go`: + +```go +type bridgeMessage struct { + subject string + data []byte + ack func() error +} + +type bridgeNATSReader interface { + Fetch(context.Context, int) ([]bridgeMessage, error) +} + +type bridgeKafkaWriter interface { + Write(context.Context, []kafka.Message) error +} + +type bridgeProcessor struct { + nats bridgeNATSReader + kafka bridgeKafkaWriter + topics map[string]string +} + +func (p bridgeProcessor) processBatch(ctx context.Context, batchSize int) error { + messages, err := p.nats.Fetch(ctx, batchSize) + if err != nil { + return err + } + kafkaMessages := make([]kafka.Message, 0, len(messages)) + for _, msg := range messages { + topic, ok := p.topics[msg.subject] + if !ok { + return fmt.Errorf("no kafka topic for nats subject %s", msg.subject) + } + key := kafkaKeyFromEnvelope(msg.data) + kafkaMessages = append(kafkaMessages, kafka.Message{Topic: topic, Key: key, Value: msg.data}) + } + if len(kafkaMessages) == 0 { + return nil + } + if err := p.kafka.Write(ctx, kafkaMessages); err != nil { + return err + } + for _, msg := range messages { + if msg.ack != nil { + if err := msg.ack(); err != nil { + return err + } + } + } + return nil +} +``` + +- [x] **Step 3: Implement real NATS and Kafka adapters** + +Use `nats.PullSubscribe`, `Fetch`, `Ack`, and `kafka.Writer.WriteMessages`. The bridge must create or reuse a durable consumer named by `NATS_CONSUMER`, default `nats-kafka-bridge`. + +- [x] **Step 4: Run bridge tests** + +Run: `go test ./cmd/nats-kafka-bridge -count=1` + +Expected: PASS. + +## Task 4: NATS Server Deployment + +**Files:** +- Create: `deploy/nats/nats-server.conf` + +- [x] **Step 1: Create NATS config** + +Create: + +```conf +server_name: lingniu-nats-01 +port: 4222 +http_port: 8222 + +jetstream { + store_dir: "/data/nats/jetstream" + max_file_store: 20GB + max_mem_store: 512MB +} +``` + +- [x] **Step 2: Deploy on Kafka ECS** + +On Kafka ECS, install or run `nats:2` with: + +```bash +mkdir -p /opt/lingniu-nats /data/nats/jetstream +docker run -d --name lingniu-nats --restart unless-stopped \ + -p 4222:4222 -p 8222:8222 \ + -v /opt/lingniu-nats/nats-server.conf:/etc/nats/nats-server.conf:ro \ + -v /data/nats:/data/nats \ + nats:2 -c /etc/nats/nats-server.conf +``` + +- [x] **Step 3: Verify NATS** + +Run from gateway ECS: + +```bash +timeout 3 bash -c '"), + NATSSubjects: splitCSV(env("NATS_STREAM_SUBJECTS", strings.Join(mapKeys(route), ","))), + KafkaBrokers: splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")), + Route: route, + BatchSize: envInt("BRIDGE_BATCH_SIZE", 500), + FetchWait: time.Duration(envInt("BRIDGE_FETCH_WAIT_MS", 1000)) * time.Millisecond, + OperationWait: time.Duration(envInt("BRIDGE_OPERATION_TIMEOUT_MS", 30000)) * time.Millisecond, + AckWait: time.Duration(envInt("NATS_ACK_WAIT_SECONDS", 60)) * time.Second, + StreamMaxAge: time.Duration(envInt("NATS_STREAM_MAX_AGE_HOURS", 24)) * time.Hour, + } +} + +type natsPullSubscription interface { + Fetch(int, ...nats.PullOpt) ([]*nats.Msg, error) +} + +type kafkaBatchWriter interface { + WriteMessages(context.Context, ...kafka.Message) error +} + +type bridgeMessage struct { + subject string + data []byte + ack func() error +} + +func runBridge(ctx context.Context, logger *slog.Logger, sub natsPullSubscription, writer kafkaBatchWriter, cfg config) { + for { + if ctx.Err() != nil { + return + } + msgs, err := sub.Fetch(cfg.BatchSize, nats.MaxWait(cfg.FetchWait)) + if err != nil { + if errors.Is(err, nats.ErrTimeout) { + continue + } + logger.Error("nats fetch failed", "error", err) + time.Sleep(time.Second) + continue + } + bridgeMessages := make([]bridgeMessage, 0, len(msgs)) + for _, msg := range msgs { + msg := msg + bridgeMessages = append(bridgeMessages, bridgeMessage{ + subject: msg.Subject, + data: msg.Data, + ack: func() error { + return msg.Ack() + }, + }) + } + operationCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), cfg.OperationWait) + err = bridgeBatch(operationCtx, writer, bridgeMessages, cfg.Route) + cancel() + if err != nil { + logger.Error("bridge batch failed", "count", len(bridgeMessages), "error", err) + continue + } + } +} + +func ensureStream(js nats.JetStreamContext, cfg config) error { + stream := &nats.StreamConfig{ + Name: cfg.NATSStream, + Subjects: cfg.NATSSubjects, + Storage: nats.FileStorage, + Retention: nats.LimitsPolicy, + MaxAge: cfg.StreamMaxAge, + Duplicates: 2 * time.Minute, + } + if _, err := js.StreamInfo(cfg.NATSStream); err == nil { + _, err = js.UpdateStream(stream) + return err + } + _, err := js.AddStream(stream) + if err == nil { + return nil + } + _, updateErr := js.UpdateStream(stream) + if updateErr == nil { + return nil + } + return err +} + +func bridgeBatch(ctx context.Context, writer kafkaBatchWriter, messages []bridgeMessage, route map[string]string) error { + if len(messages) == 0 { + return nil + } + kafkaMessages := make([]kafka.Message, 0, len(messages)) + for _, message := range messages { + topic, ok := route[message.subject] + if !ok || topic == "" { + return fmt.Errorf("kafka topic not configured for nats subject %q", message.subject) + } + kafkaMessages = append(kafkaMessages, kafkaMessage(topic, message.data)) + } + if err := writer.WriteMessages(ctx, kafkaMessages...); err != nil { + return err + } + for _, message := range messages { + if message.ack == nil { + continue + } + if err := message.ack(); err != nil { + return err + } + } + return nil +} + +func kafkaMessage(topic string, data []byte) kafka.Message { + var env envelope.FrameEnvelope + message := kafka.Message{Topic: topic, Value: data} + if err := json.Unmarshal(data, &env); err == nil { + message.Key = env.KafkaKey() + } + return message +} + +func env(key string, fallback string) string { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + return value +} + +func envInt(key string, fallback int) int { + value := strings.TrimSpace(os.Getenv(key)) + if value == "" { + return fallback + } + parsed, err := strconv.Atoi(value) + if err != nil || parsed <= 0 { + return fallback + } + return parsed +} + +func splitCSV(value string) []string { + var out []string + for _, item := range strings.Split(value, ",") { + item = strings.TrimSpace(item) + if item != "" { + out = append(out, item) + } + } + return out +} + +func mapKeys(values map[string]string) []string { + out := make([]string, 0, len(values)) + for key := range values { + out = append(out, key) + } + return out +} diff --git a/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go new file mode 100644 index 00000000..76cd47c2 --- /dev/null +++ b/go/vehicle-gateway/cmd/nats-kafka-bridge/main_test.go @@ -0,0 +1,75 @@ +package main + +import ( + "context" + "encoding/json" + "errors" + "testing" + + "github.com/segmentio/kafka-go" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestBridgeBatchWritesKafkaThenAcks(t *testing.T) { + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"} + payload, err := json.Marshal(env) + if err != nil { + t.Fatal(err) + } + writer := &recordingBridgeWriter{} + acked := 0 + + err = bridgeBatch(context.Background(), writer, []bridgeMessage{ + {subject: "vehicle.raw.jt808.v1", data: payload, ack: func() error { + acked++ + return nil + }}, + }, map[string]string{"vehicle.raw.jt808.v1": "vehicle.raw.jt808.v1"}) + if err != nil { + t.Fatalf("bridgeBatch() error = %v", err) + } + if len(writer.messages) != 1 { + t.Fatalf("kafka writes = %d, want 1", len(writer.messages)) + } + if got, want := writer.messages[0].Topic, "vehicle.raw.jt808.v1"; got != want { + t.Fatalf("topic = %q, want %q", got, want) + } + if got, want := string(writer.messages[0].Key), "JT808:13307795425"; got != want { + t.Fatalf("key = %q, want %q", got, want) + } + if acked != 1 { + t.Fatalf("acks = %d, want 1", acked) + } +} + +func TestBridgeBatchDoesNotAckWhenKafkaFails(t *testing.T) { + writer := &recordingBridgeWriter{err: errors.New("kafka unavailable")} + acked := 0 + + err := bridgeBatch(context.Background(), writer, []bridgeMessage{ + {subject: "vehicle.event.unified.v1", data: []byte(`{"protocol":"JT808"}`), ack: func() error { + acked++ + return nil + }}, + }, map[string]string{"vehicle.event.unified.v1": "vehicle.event.unified.v1"}) + if err == nil { + t.Fatal("bridgeBatch() error is nil, want kafka error") + } + if acked != 0 { + t.Fatalf("acks = %d, want 0", acked) + } +} + +type recordingBridgeWriter struct { + messages []kafka.Message + err error +} + +func (w *recordingBridgeWriter) WriteMessages(_ context.Context, messages ...kafka.Message) error { + if w.err != nil { + return w.err + } + w.messages = append(w.messages, messages...) + return nil +} diff --git a/go/vehicle-gateway/go.mod b/go/vehicle-gateway/go.mod index 9be8d5bc..ae277b33 100644 --- a/go/vehicle-gateway/go.mod +++ b/go/vehicle-gateway/go.mod @@ -7,10 +7,11 @@ require ( github.com/alicebob/miniredis/v2 v2.35.0 github.com/eclipse/paho.mqtt.golang v1.5.1 github.com/go-sql-driver/mysql v1.9.3 + github.com/nats-io/nats.go v1.52.0 github.com/redis/go-redis/v9 v9.17.2 github.com/segmentio/kafka-go v0.4.49 github.com/taosdata/driver-go/v3 v3.8.1 - golang.org/x/text v0.29.0 + golang.org/x/text v0.35.0 ) require ( @@ -20,11 +21,15 @@ require ( github.com/google/uuid v1.6.0 // indirect github.com/gorilla/websocket v1.5.3 // indirect github.com/json-iterator/go v1.1.12 // indirect - github.com/klauspost/compress v1.15.9 // indirect + github.com/klauspost/compress v1.18.5 // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/nats-io/nkeys v0.4.15 // indirect + github.com/nats-io/nuid v1.0.1 // indirect github.com/pierrec/lz4/v4 v4.1.15 // indirect github.com/yuin/gopher-lua v1.1.1 // indirect - golang.org/x/net v0.44.0 // indirect - golang.org/x/sync v0.17.0 // indirect + golang.org/x/crypto v0.49.0 // indirect + golang.org/x/net v0.51.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.42.0 // indirect ) diff --git a/go/vehicle-gateway/go.sum b/go/vehicle-gateway/go.sum index 1dafdbf9..baf711b1 100644 --- a/go/vehicle-gateway/go.sum +++ b/go/vehicle-gateway/go.sum @@ -28,12 +28,18 @@ github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/ad github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/sqlstruct v0.0.0-20201105191214-5f3e10d3ab46/go.mod h1:yyMNCyc/Ib3bDTKd379tNMpB/7/H5TjM2Y9QJ5THLbE= -github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY= -github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU= +github.com/klauspost/compress v1.18.5 h1:/h1gH5Ce+VWNLSWqPzOVn6XBO+vJbCNGvjoaGBFW2IE= +github.com/klauspost/compress v1.18.5/go.mod h1:cwPg85FWrGar70rWktvGQj8/hthj3wpl0PGDogxkrSQ= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/nats-io/nats.go v1.52.0 h1:n3avV4VBsCgsdwh71TppsTwtv+QdPs7ntSKM8qJLGsc= +github.com/nats-io/nats.go v1.52.0/go.mod h1:26HypzazeOkyO3/mqd1zZd53STJN0EjCYF9Uy2ZOBno= +github.com/nats-io/nkeys v0.4.15 h1:JACV5jRVO9V856KOapQ7x+EY8Jo3qw1vJt/9Jpwzkk4= +github.com/nats-io/nkeys v0.4.15/go.mod h1:CpMchTXC9fxA5zrMo4KpySxNjiDVvr8ANOSZdiNfUrs= +github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= +github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0= github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= @@ -61,12 +67,16 @@ github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6 github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM= github.com/yuin/gopher-lua v1.1.1 h1:kYKnWBjvbNP4XLT3+bPEwAXJx262OhaHDWDVOPjL46M= github.com/yuin/gopher-lua v1.1.1/go.mod h1:GBR0iDaNXjAgGg9zfCvksxSRnQx76gclCIb7kdAd1Pw= -golang.org/x/net v0.44.0 h1:evd8IRDyfNBMBTTY5XRF1vaZlD+EmWx6x8PkhR04H/I= -golang.org/x/net v0.44.0/go.mod h1:ECOoLqd5U3Lhyeyo/QDCEVQ4sNgYsqvCZ722XogGieY= -golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug= -golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= -golang.org/x/text v0.29.0 h1:1neNs90w9YzJ9BocxfsQNHKuAT4pkghyXc4nhZ6sJvk= -golang.org/x/text v0.29.0/go.mod h1:7MhJOA9CD2qZyOKYazxdYMF85OwPdEr9jTtBpO7ydH4= +golang.org/x/crypto v0.49.0 h1:+Ng2ULVvLHnJ/ZFEq4KdcDd/cfjrrjjNSXNzxg0Y4U4= +golang.org/x/crypto v0.49.0/go.mod h1:ErX4dUh2UM+CFYiXZRTcMpEcN8b/1gxEuv3nODoYtCA= +golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo= +golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= +golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= diff --git a/go/vehicle-gateway/internal/eventbus/nats_sink.go b/go/vehicle-gateway/internal/eventbus/nats_sink.go new file mode 100644 index 00000000..21d7153b --- /dev/null +++ b/go/vehicle-gateway/internal/eventbus/nats_sink.go @@ -0,0 +1,117 @@ +package eventbus + +import ( + "context" + "errors" + "fmt" + "time" + + "github.com/nats-io/nats.go" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type NATSConfig struct { + URL string + Name string + RawSubjects map[envelope.Protocol]string + UnifiedSubject string +} + +type NATSSink struct { + conn *nats.Conn + publisher natsPublisher + rawSubjects map[envelope.Protocol]string + unifiedSubject string +} + +type NATSPublishOption = nats.PubOpt + +type natsPublisher interface { + Publish(context.Context, string, []byte, ...NATSPublishOption) error +} + +func NewNATSSink(cfg NATSConfig) (*NATSSink, error) { + if cfg.URL == "" { + return nil, errors.New("nats url is required") + } + name := cfg.Name + if name == "" { + name = "lingniu-vehicle-gateway" + } + conn, err := nats.Connect(cfg.URL, nats.Name(name), nats.Timeout(5*time.Second)) + if err != nil { + return nil, err + } + js, err := conn.JetStream() + if err != nil { + conn.Close() + return nil, err + } + sink := newNATSSinkWithPublisher(natsJetStreamPublisher{js: js}, cfg) + sink.conn = conn + return sink, nil +} + +func newNATSSinkWithPublisher(publisher natsPublisher, cfg NATSConfig) *NATSSink { + rawSubjects := map[envelope.Protocol]string{ + envelope.ProtocolGB32960: "vehicle.raw.gb32960.v1", + envelope.ProtocolJT808: "vehicle.raw.jt808.v1", + envelope.ProtocolYutongMQTT: "vehicle.raw.yutong-mqtt.v1", + } + for protocol, subject := range cfg.RawSubjects { + if subject != "" { + rawSubjects[protocol] = subject + } + } + unifiedSubject := cfg.UnifiedSubject + if unifiedSubject == "" { + unifiedSubject = "vehicle.event.unified.v1" + } + return &NATSSink{ + publisher: publisher, + rawSubjects: rawSubjects, + unifiedSubject: unifiedSubject, + } +} + +func (s *NATSSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error { + subject, ok := s.rawSubjects[env.Protocol] + if !ok || subject == "" { + return fmt.Errorf("raw subject not configured for protocol %s", env.Protocol) + } + return s.publish(ctx, subject, env) +} + +func (s *NATSSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error { + if s.unifiedSubject == "" { + return errors.New("unified subject is empty") + } + return s.publish(ctx, s.unifiedSubject, env) +} + +func (s *NATSSink) Close() error { + if s == nil || s.conn == nil { + return nil + } + _ = s.conn.Drain() + s.conn.Close() + return nil +} + +func (s *NATSSink) publish(ctx context.Context, subject string, env envelope.FrameEnvelope) error { + payload, err := env.MarshalJSONBytes() + if err != nil { + return err + } + return s.publisher.Publish(ctx, subject, payload, nats.MsgId(env.StableEventID())) +} + +type natsJetStreamPublisher struct { + js nats.JetStreamContext +} + +func (p natsJetStreamPublisher) Publish(ctx context.Context, subject string, data []byte, opts ...NATSPublishOption) error { + _, err := p.js.Publish(subject, data, append(opts, nats.Context(ctx))...) + return err +} diff --git a/go/vehicle-gateway/internal/eventbus/nats_sink_test.go b/go/vehicle-gateway/internal/eventbus/nats_sink_test.go new file mode 100644 index 00000000..36938d41 --- /dev/null +++ b/go/vehicle-gateway/internal/eventbus/nats_sink_test.go @@ -0,0 +1,55 @@ +package eventbus + +import ( + "context" + "encoding/json" + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestNATSSinkRoutesRawAndUnifiedSubjects(t *testing.T) { + publisher := &recordingNATSPublisher{} + sink := newNATSSinkWithPublisher(publisher, NATSConfig{ + RawSubjects: map[envelope.Protocol]string{ + envelope.ProtocolJT808: "vehicle.raw.jt808.v1", + }, + UnifiedSubject: "vehicle.event.unified.v1", + }) + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"} + + if err := sink.PublishRaw(context.Background(), env); err != nil { + t.Fatalf("PublishRaw() error = %v", err) + } + if err := sink.PublishUnified(context.Background(), env); err != nil { + t.Fatalf("PublishUnified() error = %v", err) + } + + if got, want := publisher.messages[0].subject, "vehicle.raw.jt808.v1"; got != want { + t.Fatalf("raw subject = %q, want %q", got, want) + } + if got, want := publisher.messages[1].subject, "vehicle.event.unified.v1"; got != want { + t.Fatalf("unified subject = %q, want %q", got, want) + } + var decoded envelope.FrameEnvelope + if err := json.Unmarshal(publisher.messages[0].data, &decoded); err != nil { + t.Fatalf("raw payload is not envelope json: %v", err) + } + if decoded.Phone != "13307795425" { + t.Fatalf("decoded phone = %q", decoded.Phone) + } +} + +type recordingNATSPublisher struct { + messages []recordedNATSMessage +} + +type recordedNATSMessage struct { + subject string + data []byte +} + +func (p *recordingNATSPublisher) Publish(_ context.Context, subject string, data []byte, _ ...NATSPublishOption) error { + p.messages = append(p.messages, recordedNATSMessage{subject: subject, data: append([]byte(nil), data...)}) + return nil +}