309 lines
12 KiB
Go
309 lines
12 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
"log/slog"
|
|
"os"
|
|
"os/signal"
|
|
"strings"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
_ "github.com/go-sql-driver/mysql"
|
|
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/health"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808"
|
|
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics"
|
|
)
|
|
|
|
func main() {
|
|
logger := observability.NewLogger("vehicle-gateway")
|
|
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
|
defer stop()
|
|
|
|
sink, err := buildSink(ctx, logger)
|
|
if err != nil {
|
|
logger.Error("build sink failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
defer sink.Close()
|
|
resolver, closeResolver, err := buildIdentityResolver(ctx, logger)
|
|
if err != nil {
|
|
logger.Error("build identity resolver failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
defer closeResolver()
|
|
registry := metrics.NewRegistry()
|
|
health.Start(ctx, logger, health.NewServer(env("HEALTH_ADDR", ""), "vehicle-gateway", nil, registry))
|
|
publishUnified := envBool("PUBLISH_UNIFIED_ENABLED", false)
|
|
|
|
protocols := []gateway.TCPProtocol{
|
|
{
|
|
Protocol: envelope.ProtocolGB32960,
|
|
Addr: env("GB32960_TCP_ADDR", ":32960"),
|
|
Extract: gb32960.ExtractFrames,
|
|
Parse: gb32960.ParseFrame,
|
|
Respond: gb32960.AutoResponse,
|
|
},
|
|
{
|
|
Protocol: envelope.ProtocolJT808,
|
|
Addr: env("JT808_TCP_ADDR", ":808"),
|
|
Extract: jt808.ExtractFrames,
|
|
Parse: jt808.ParseFrame,
|
|
Respond: jt808.NewAutoResponder(env("JT808_REGISTER_AUTH_CODE", "g7gps")).Respond,
|
|
},
|
|
}
|
|
|
|
var wg sync.WaitGroup
|
|
errs := make(chan error, len(protocols))
|
|
for _, protocol := range protocols {
|
|
server, err := gateway.NewTCPServer(gateway.TCPServerConfig{
|
|
Protocol: protocol,
|
|
Sink: sink,
|
|
Resolver: resolver,
|
|
Logger: logger,
|
|
Metrics: registry,
|
|
ReadBufferSize: envInt("TCP_READ_BUFFER_BYTES", 64*1024),
|
|
IdleTimeout: time.Duration(envInt("TCP_IDLE_TIMEOUT_SECONDS", 180)) * time.Second,
|
|
MaxConnections: envInt("TCP_MAX_CONNECTIONS", 120_000),
|
|
PublishUnified: publishUnified,
|
|
})
|
|
if err != nil {
|
|
logger.Error("build tcp server failed", "protocol", protocol.Protocol, "error", err)
|
|
os.Exit(1)
|
|
}
|
|
wg.Add(1)
|
|
go func() {
|
|
defer wg.Done()
|
|
if err := server.ListenAndServe(ctx); err != nil {
|
|
errs <- err
|
|
}
|
|
}()
|
|
}
|
|
if envBool("YUTONG_MQTT_ENABLED", false) {
|
|
client, err := gateway.NewMQTTClient(gateway.MQTTClientConfig{
|
|
EndpointName: env("YUTONG_MQTT_ENDPOINT", env("YUTONG_MQTT_ENDPOINT_NAME", "yutong")),
|
|
Broker: env("YUTONG_MQTT_URI", ""),
|
|
ClientID: env("YUTONG_MQTT_CLIENT_ID", "lingniu-go-yutong-mqtt"),
|
|
Username: env("YUTONG_MQTT_USERNAME", ""),
|
|
Password: env("YUTONG_MQTT_PASSWORD", ""),
|
|
Topics: splitCSV(env("YUTONG_MQTT_TOPICS", env("YUTONG_MQTT_TOPIC", "/ytforward/shln/+"))),
|
|
QoS: byte(envInt("YUTONG_MQTT_QOS", 2)),
|
|
CleanSession: envBool("YUTONG_MQTT_CLEAN_SESSION", false),
|
|
KeepAlive: time.Duration(envInt("YUTONG_MQTT_KEEP_ALIVE_SECONDS", 20)) * time.Second,
|
|
ConnectTimeout: time.Duration(envInt("YUTONG_MQTT_CONNECTION_TIMEOUT_SECONDS", 10)) * time.Second,
|
|
TLSCACertPath: env("YUTONG_MQTT_TLS_CA_PEM", ""),
|
|
TLSClientCertPath: env("YUTONG_MQTT_TLS_CLIENT_PEM", ""),
|
|
TLSClientKeyPath: env("YUTONG_MQTT_TLS_CLIENT_KEY", ""),
|
|
TLSHostnameVerification: envBool("YUTONG_MQTT_TLS_HOSTNAME_VERIFICATION_ENABLED", true),
|
|
Sink: sink,
|
|
Resolver: resolver,
|
|
Logger: logger,
|
|
Metrics: registry,
|
|
PublishUnified: publishUnified,
|
|
})
|
|
if err != nil {
|
|
logger.Error("build yutong mqtt client failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
if err := client.Start(ctx); err != nil {
|
|
logger.Error("start yutong mqtt client failed", "error", err)
|
|
os.Exit(1)
|
|
}
|
|
logger.Info("yutong mqtt client started")
|
|
}
|
|
|
|
logger.Info("vehicle gateway started")
|
|
select {
|
|
case <-ctx.Done():
|
|
case err := <-errs:
|
|
logger.Error("vehicle gateway listener failed", "error", err)
|
|
stop()
|
|
}
|
|
wg.Wait()
|
|
}
|
|
|
|
func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.Resolver, func(), error) {
|
|
dsn := env("IDENTITY_MYSQL_DSN", strings.TrimSpace(os.Getenv("MYSQL_DSN")))
|
|
if strings.TrimSpace(dsn) == "" {
|
|
logger.Warn("identity mysql dsn is empty; using noop identity resolver")
|
|
return identity.NoopResolver{}, func() {}, nil
|
|
}
|
|
db, err := sql.Open("mysql", dsn)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if err := db.PingContext(ctx); err != nil {
|
|
_ = db.Close()
|
|
return nil, nil, err
|
|
}
|
|
table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding")
|
|
resolver := identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{
|
|
LocationTouchInterval: time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second,
|
|
LookupCacheTTL: time.Duration(envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600)) * time.Second,
|
|
})
|
|
if envBool("IDENTITY_MYSQL_ENSURE_SCHEMA", true) {
|
|
if err := resolver.EnsureSchema(ctx); err != nil {
|
|
_ = db.Close()
|
|
return nil, nil, err
|
|
}
|
|
}
|
|
logger.Info("identity mysql resolver enabled", "table", table, "lookup_cache_ttl_seconds", envInt("IDENTITY_LOOKUP_CACHE_TTL_SECONDS", 600))
|
|
return resolver, func() { _ = db.Close() }, nil
|
|
}
|
|
|
|
func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) {
|
|
if strings.TrimSpace(os.Getenv("NATS_URL")) != "" {
|
|
sink, err := eventbus.NewNATSSink(natsSinkConfigFromEnv())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var out eventbus.Sink = eventbus.NewRetryingSink(sink, eventbus.RetryConfig{
|
|
Attempts: envInt("NATS_PUBLISH_ATTEMPTS", 3),
|
|
Backoff: time.Duration(envInt("NATS_PUBLISH_BACKOFF_MS", 100)) * time.Millisecond,
|
|
AttemptTimeout: time.Duration(envInt("NATS_PUBLISH_TIMEOUT_MS", 3000)) * time.Millisecond,
|
|
})
|
|
if envBool("NATS_ASYNC_ENABLED", true) {
|
|
queueSize := envInt("NATS_ASYNC_QUEUE_SIZE", envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000))
|
|
workers := envInt("NATS_ASYNC_WORKERS", envInt("KAFKA_ASYNC_WORKERS", 8))
|
|
timeout := time.Duration(envInt("NATS_ASYNC_PUBLISH_TIMEOUT_MS", envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000))) * time.Millisecond
|
|
out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{
|
|
QueueSize: queueSize,
|
|
Workers: workers,
|
|
OperationTimeout: timeout,
|
|
OnError: func(err error) {
|
|
logger.Warn("nats async publish failed", "error", err)
|
|
},
|
|
})
|
|
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), "publish_fields_enabled", true)
|
|
return out, nil
|
|
}
|
|
brokers := splitCSV(os.Getenv("KAFKA_BROKERS"))
|
|
if len(brokers) == 0 {
|
|
logger.Warn("KAFKA_BROKERS is empty; using log sink")
|
|
return eventbus.NewLogSink(logger), nil
|
|
}
|
|
sink, err := eventbus.NewKafkaSink(eventbus.KafkaConfig{
|
|
Brokers: brokers,
|
|
RawTopics: map[envelope.Protocol]string{
|
|
envelope.ProtocolGB32960: env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960),
|
|
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 {
|
|
return nil, err
|
|
}
|
|
var out eventbus.Sink = eventbus.NewRetryingSink(sink, eventbus.RetryConfig{
|
|
Attempts: envInt("KAFKA_PUBLISH_ATTEMPTS", 3),
|
|
Backoff: time.Duration(envInt("KAFKA_PUBLISH_BACKOFF_MS", 100)) * time.Millisecond,
|
|
AttemptTimeout: time.Duration(envInt("KAFKA_PUBLISH_TIMEOUT_MS", 3000)) * time.Millisecond,
|
|
})
|
|
spoolDir := strings.TrimSpace(os.Getenv("KAFKA_SPOOL_DIR"))
|
|
if spoolDir != "" {
|
|
replayBatchSize := envInt("KAFKA_SPOOL_REPLAY_BATCH_SIZE", 200)
|
|
durable := eventbus.NewDurableSink(out, eventbus.DurableConfig{Directory: spoolDir, ReplayBatchSize: replayBatchSize})
|
|
interval := time.Duration(envInt("KAFKA_SPOOL_REPLAY_INTERVAL_MS", 1000)) * time.Millisecond
|
|
go durable.ReplayLoop(ctx, interval, func(err error) {
|
|
logger.Warn("kafka spool replay failed", "error", err)
|
|
})
|
|
logger.Info("kafka durable spool enabled", "dir", spoolDir, "replay_interval_ms", interval.Milliseconds(), "replay_batch_size", replayBatchSize)
|
|
out = durable
|
|
}
|
|
if envBool("KAFKA_ASYNC_ENABLED", true) {
|
|
queueSize := envInt("KAFKA_ASYNC_QUEUE_SIZE", 100000)
|
|
workers := envInt("KAFKA_ASYNC_WORKERS", 8)
|
|
timeout := time.Duration(envInt("KAFKA_ASYNC_PUBLISH_TIMEOUT_MS", 30000)) * time.Millisecond
|
|
out = eventbus.NewAsyncSink(out, eventbus.AsyncConfig{
|
|
QueueSize: queueSize,
|
|
Workers: workers,
|
|
OperationTimeout: timeout,
|
|
OnError: func(err error) {
|
|
logger.Warn("kafka async publish failed", "error", err)
|
|
},
|
|
})
|
|
logger.Info("kafka async publish enabled", "queue_size", queueSize, "workers", workers, "publish_timeout_ms", timeout.Milliseconds())
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
func natsSinkConfigFromEnv() eventbus.NATSConfig {
|
|
return eventbus.NATSConfig{
|
|
URL: env("NATS_URL", ""),
|
|
Name: env("NATS_CLIENT_NAME", "lingniu-vehicle-gateway"),
|
|
RawSubjects: map[envelope.Protocol]string{
|
|
envelope.ProtocolGB32960: env("NATS_SUBJECT_GB32960_RAW", env("KAFKA_TOPIC_GB32960_RAW", topics.RawGB32960)),
|
|
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)),
|
|
}
|
|
}
|
|
|
|
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
|
|
}
|
|
var out int
|
|
for _, r := range value {
|
|
if r < '0' || r > '9' {
|
|
return fallback
|
|
}
|
|
out = out*10 + int(r-'0')
|
|
}
|
|
if out <= 0 {
|
|
return fallback
|
|
}
|
|
return out
|
|
}
|
|
|
|
func envBool(key string, fallback bool) bool {
|
|
value := strings.ToLower(strings.TrimSpace(os.Getenv(key)))
|
|
if value == "" {
|
|
return fallback
|
|
}
|
|
return value == "1" || value == "true" || value == "yes" || value == "on"
|
|
}
|
|
|
|
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
|
|
}
|