lingniu-vehicle-ingest 模块与数据流

flowchart TB
  subgraph external["外部来源"]
    vehicle["车辆终端
GB/T 32960 / JT808 / JT1078"]
    mqttSource["车企或平台 MQTT"]
    commandClient["业务系统 / 运维平台
HTTP 下行命令"]
  end

  subgraph protocolModules["协议模块 modules/protocols"]
    gb["protocol-gb32960
32960 Netty 接入、鉴权、ACK、解析"]
    jt808["protocol-jt808
808 Netty 接入、会话、位置/批量位置、媒体、透传、下行"]
    jt1078["protocol-jt1078
1078 信令 + TCP/UDP RTP 媒体流归档"]
    jsatl12["protocol-jsatl12
主动安全附件流接入"]
  end

  subgraph inbound["入口适配 modules/inbound"]
    mqtt["inbound-mqtt
MQTT endpoint 生命周期、PEM TLS"]
    mqttProfile["MqttProfileRegistry
按 endpoint profile 解析/映射"]
  end

  subgraph protocolBase["核心与公共能力 modules/core"]
    codec["ingest-codec-common
BCD / CRC / BCC / bit 工具"]
    session["session-core
设备会话、命令分发接口、会话存储"]
    identity["vehicle-identity
跨协议身份解析、MySQL 绑定表"]
    obs["observability
metrics / health"]
    api["ingest-api
ProtocolId / RawFrame / VehicleEvent / 注解 SPI"]
    registry["HandlerRegistry
按协议、命令、infoType 路由"]
    dispatcher["Dispatcher
RawFrame 到 Handler 到 EventBus"]
    bus["DisruptorEventBus
高吞吐事件发布"]
  end

  subgraph sink["输出与明细存储 modules/sinks"]
    kafka["sink-kafka
VehicleEvent 到 Protobuf Envelope 到 Kafka"]
    archive["sink-archive
RawArchive 到本地文件系统冷存"]
    tdengineStore["tdengine-history-store
TDengine raw_frames + locations"]
  end

  subgraph services["消费服务 modules/services"]
    history["event-history-service
Kafka event/raw 到 TDengine 历史查询"]
    state["vehicle-state-service
可选 Redis 热状态
optional-latest-state"]
    stat["vehicle-stat-service
Kafka 全字段事件到指标表"]
  end

  subgraph apps["应用入口 modules/apps"]
    gbApp["gb32960-ingest-app
GB32960 TCP 接入"]
    jtApp["jt808-ingest-app
JT808 TCP 接入"]
    yutongApp["yutong-mqtt-app
宇通 MQTT 接入"]
    historyApp["vehicle-history-app
历史查询与 TDengine 写入"]
    analyticsApp["vehicle-analytics-app
808 日里程指标消费"]
    gateway["command-gateway
可选 HTTP 设备命令入口"]
  end

  vehicle --> gbApp
  vehicle --> jtApp
  vehicle --> jt1078
  vehicle --> jsatl12
  mqttSource --> yutongApp
  gbApp --> gb
  jtApp --> jt808
  yutongApp --> mqtt
  mqtt --> mqttProfile
  mqttProfile --> api
  commandClient --> gateway

  gb --> codec
  jt808 --> codec
  jt1078 --> codec
  jsatl12 --> codec

  gb --> api
  jt808 --> api
  jsatl12 --> archive
  jsatl12 --> jt808

  api --> registry
  registry --> dispatcher
  dispatcher --> bus
  bus --> kafka
  bus --> archive
  kafka --> historyApp
  kafka --> analyticsApp
  historyApp --> history
  analyticsApp --> stat
  kafka -.可选热状态.-> state
  history --> tdengineStore

  gateway --> session
  session --> jt808
  session --> gb
  identity --> jt808
  identity --> mqttProfile

  obs -.监控.-> dispatcher
  obs -.监控.-> bus
  obs -.监控.-> kafka

  classDef entry fill:#eff6ff,stroke:#2563eb,color:#1e3a8a;
  classDef core fill:#ecfdf3,stroke:#0f766e,color:#134e48;
  classDef sink fill:#fffbeb,stroke:#b45309,color:#7c2d12;
  classDef command fill:#f5f3ff,stroke:#6d28d9,color:#4c1d95;
  classDef support fill:#f8fafc,stroke:#64748b,color:#334155;

  class gb,jt808,jt1078,jsatl12,mqtt,mqttProfile,gbApp,jtApp,yutongApp,historyApp,analyticsApp entry;
  class api,registry,dispatcher,bus,identity core;
  class kafka,archive,tdengineStore sink;
  class gateway,session command;
  class codec,obs support;
      

sequenceDiagram
  autonumber
  participant Device as 车辆/平台
  participant Inbound as Inbound Adapter
Netty/MQTT
  participant Decoder as FrameDecoder
MessageDecoder
  participant Dispatcher as Dispatcher
  participant Handler as Protocol Handler
Mapper
  participant EventBus as DisruptorEventBus
  participant Kafka as sink-kafka
Kafka Envelope
  participant Archive as sink-archive
ArchiveStore
  participant History as vehicle-history-app
  participant TDengine as tdengine-history-store
raw_frames + locations
  participant Consumer as 下游业务消费者

  Device->>Inbound: 原始报文 bytes / MQTT message
  Inbound->>Decoder: 粘包拆帧、校验、协议体解析
  Inbound->>Inbound: MQTT 按 endpoint profile 选择厂商解析器
  Decoder->>Dispatcher: RawFrame(protocol, command, infoType, payload, rawBytes)
  Dispatcher->>EventBus: 先发布 RawArchive 事件
  EventBus->>Archive: 保存原始 bytes，生成可回放材料
  Dispatcher->>Handler: 根据 protocol + command + infoType 路由
  Handler->>Handler: 协议字段映射为内部字段
  Dispatcher->>Handler: 绑定 rawArchiveKey / rawArchiveUri
  Handler->>EventBus: VehicleEvent(Realtime / Location / Alarm / Login ...)
  EventBus->>Kafka: 投递规整后的业务事件
  Kafka->>Kafka: EnvelopeMapper 转 Protobuf
  Kafka->>History: event/raw topic，key=vin，单车有序
  History->>TDengine: 写 raw_frames、locations 和 raw parsed JSON
  TDengine->>Consumer: 分页历史查询、原始帧追溯、位置查询
  Kafka->>Consumer: Redis 热状态、指标统计、告警工单等下游消费
      

flowchart LR
  terminal["氢能车 TBOX
GB/T 32960"] --> netty["Gb32960NettyServer
TCP / TLS / Idle 检测"]
  netty --> access["Gb32960AccessService
VIN 白名单 / 平台登录鉴权"]
  netty --> frame["Gb32960FrameDecoder
拆帧、转义、校验"]
  frame --> decoder["Gb32960MessageDecoder
Header + Body 解析"]
  decoder --> diag["Gb32960FrameDiagnostics
首帧、Raw 块、异常诊断"]
  decoder --> handler["Gb32960ChannelHandler
ACK / NACK / dispatch"]
  handler --> ack["Gb32960AckService
登录应答、失败断开"]
  handler --> dispatcher["Dispatcher"]
  dispatcher --> rtHandler["Gb32960RealtimeHandler"]
  rtHandler --> mapper["Gb32960EventMapper
协议字段到内部字段"]
  mapper --> realtime["RealtimePayload
速度、里程、电量、氢量、压力、温度"]
  mapper --> alarm["AlarmPayload
安全分类、氢泄露、告警等级"]
  mapper --> login["Login / Logout / Heartbeat"]
  realtime --> bus["DisruptorEventBus"]
  alarm --> bus
  login --> bus
  bus --> kafka["Kafka Envelope"]
  bus --> archive["RawArchive 冷存"]

  classDef safety fill:#fff1f3,stroke:#b42318,color:#7a271a;
  classDef core fill:#ecfdf3,stroke:#0f766e,color:#134e48;
  classDef entry fill:#eff6ff,stroke:#2563eb,color:#1e3a8a;
  classDef sink fill:#fffbeb,stroke:#b45309,color:#7c2d12;

  class terminal,netty,frame,decoder,handler entry;
  class access,diag,ack,dispatcher,rtHandler,mapper,bus core;
  class alarm safety;
  class kafka,archive sink;