95 lines
2.3 KiB
Go
95 lines
2.3 KiB
Go
package eventbus
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
|
|
"github.com/segmentio/kafka-go"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
)
|
|
|
|
type Sink interface {
|
|
PublishRaw(context.Context, envelope.FrameEnvelope) error
|
|
PublishUnified(context.Context, envelope.FrameEnvelope) error
|
|
Close() error
|
|
}
|
|
|
|
type KafkaSink struct {
|
|
writer kafkaWriter
|
|
rawTopics map[envelope.Protocol]string
|
|
unifiedTopic string
|
|
}
|
|
|
|
type kafkaWriter interface {
|
|
WriteMessages(context.Context, ...kafka.Message) error
|
|
Close() error
|
|
}
|
|
|
|
type KafkaConfig struct {
|
|
Brokers []string
|
|
RawTopics map[envelope.Protocol]string
|
|
UnifiedTopic string
|
|
}
|
|
|
|
func NewKafkaSink(cfg KafkaConfig) (*KafkaSink, error) {
|
|
if len(cfg.Brokers) == 0 {
|
|
return nil, errors.New("kafka brokers are required")
|
|
}
|
|
return newKafkaSinkWithWriter(&kafka.Writer{
|
|
Addr: kafka.TCP(cfg.Brokers...),
|
|
Balancer: &kafka.Hash{},
|
|
AllowAutoTopicCreation: false,
|
|
}, cfg), nil
|
|
}
|
|
|
|
func newKafkaSinkWithWriter(writer kafkaWriter, cfg KafkaConfig) *KafkaSink {
|
|
rawTopics := 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, topic := range cfg.RawTopics {
|
|
if topic != "" {
|
|
rawTopics[protocol] = topic
|
|
}
|
|
}
|
|
unifiedTopic := cfg.UnifiedTopic
|
|
if unifiedTopic == "" {
|
|
unifiedTopic = "vehicle.event.unified.v1"
|
|
}
|
|
return &KafkaSink{writer: writer, rawTopics: rawTopics, unifiedTopic: unifiedTopic}
|
|
}
|
|
|
|
func (s *KafkaSink) PublishRaw(ctx context.Context, env envelope.FrameEnvelope) error {
|
|
topic, ok := s.rawTopics[env.Protocol]
|
|
if !ok || topic == "" {
|
|
return fmt.Errorf("raw topic not configured for protocol %s", env.Protocol)
|
|
}
|
|
return s.publish(ctx, topic, env)
|
|
}
|
|
|
|
func (s *KafkaSink) PublishUnified(ctx context.Context, env envelope.FrameEnvelope) error {
|
|
return s.publish(ctx, s.unifiedTopic, env)
|
|
}
|
|
|
|
func (s *KafkaSink) Close() error {
|
|
if s == nil || s.writer == nil {
|
|
return nil
|
|
}
|
|
return s.writer.Close()
|
|
}
|
|
|
|
func (s *KafkaSink) publish(ctx context.Context, topic string, env envelope.FrameEnvelope) error {
|
|
payload, err := env.MarshalJSONBytes()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return s.writer.WriteMessages(ctx, kafka.Message{
|
|
Topic: topic,
|
|
Key: env.KafkaKey(),
|
|
Value: payload,
|
|
})
|
|
}
|