shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: write protocol acknowledgements from go gateway

Browse Source
This commit is contained in:
lingniu
2026-07-02 00:54:31 +08:00
parent 055373c405
commit 6d1d0aa85e
7 changed files with 376 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
go/vehicle-gateway/internal/gateway/tcp_server.go
View File

@@ -21,11 +21,14 @@ type FrameExtractor func([]byte) (frames [][]byte, remainder []byte, err error)
type FrameParser func(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope.FrameEnvelope, error)
type FrameResponder func(raw []byte, env envelope.FrameEnvelope) (response []byte, ok bool, err error)
type TCPProtocol struct {
Protocol envelope.Protocol
Addr string
Extract FrameExtractor
Parse FrameParser
Respond FrameResponder
}
type TCPServer struct {
@@ -153,7 +156,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) {
}
pending = remainder
for _, frame := range frames {
s.handleFrame(ctx, frame, source)
s.handleFrame(ctx, conn, frame, source)
}
}
if err != nil {
@@ -174,7 +177,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) {
}
}
func (s *TCPServer) handleFrame(ctx context.Context, raw []byte, source string) {
func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string) {
receivedAtMS := time.Now().UnixMilli()
env, err := s.protocol.Parse(raw, receivedAtMS, source)
if err != nil {
@@ -213,6 +216,21 @@ func (s *TCPServer) handleFrame(ctx context.Context, raw []byte, source string)
s.logger.Error("publish unified failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err)
return
}
if s.protocol.Respond == nil {
return
}
response, ok, err := s.protocol.Respond(raw, env)
if err != nil {
s.logger.Warn("build protocol response failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err)
return
}
if !ok || len(response) == 0 {
return
}
_ = conn.SetWriteDeadline(time.Now().Add(5 * time.Second))
if _, err := conn.Write(response); err != nil {
s.logger.Warn("write protocol response failed", "protocol", s.protocol.Protocol, "event_id", env.StableEventID(), "error", err)
}
}
func (p TCPProtocol) String() string {

32
go/vehicle-gateway/internal/gateway/tcp_server_test.go
View File

@@ -3,6 +3,7 @@ package gateway
import (
"context"
"encoding/hex"
"io"
"log/slog"
"net"
"testing"
@@ -73,6 +74,37 @@ func TestTCPServerPublishesBadFrameOnlyToRaw(t *testing.T) {
}
}
func TestTCPServerWritesProtocolResponseAfterPublish(t *testing.T) {
frame := buildGBFrame(0x07, 0xfe, "LNBSCB3D4R1234567", nil)
sink := &recordingSink{}
server := newTestServer(t, TCPProtocol{
Protocol: envelope.ProtocolGB32960,
Addr: ":0",
Extract: gb32960.ExtractFrames,
Parse: gb32960.ParseFrame,
Respond: func(_ []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
if len(sink.unified) != 1 || sink.unified[0].EventID != env.EventID {
t.Fatalf("response built before publish: raw=%d unified=%d", len(sink.raw), len(sink.unified))
}
return []byte("ACK"), true, nil
},
}, sink)
client, done := runPipe(t, server)
if _, err := client.Write(frame); err != nil {
t.Fatalf("client.Write() error = %v", err)
}
buf := make([]byte, 3)
if _, err := io.ReadFull(client, buf); err != nil {
t.Fatalf("read response error = %v", err)
}
if string(buf) != "ACK" {
t.Fatalf("response = %q", string(buf))
}
_ = client.Close()
<-done
}
func newTestServer(t *testing.T, protocol TCPProtocol, sink *recordingSink) *TCPServer {
t.Helper()
server, err := NewTCPServer(TCPServerConfig{

32
go/vehicle-gateway/internal/protocol/gb32960/parser_test.go
View File

@@ -90,6 +90,38 @@ func TestExtractFramesSupportsExtendedPlatformLogin(t *testing.T) {
}
}
func TestAutoResponseEchoesRawVINAndOriginalTime(t *testing.T) {
body := []byte{0x1a, 0x07, 0x02, 0x00, 0x26, 0x0f, 0x00, 0x01}
body = append(body, fixedASCII("Hyundai", 12)...)
body = append(body, fixedASCII("ack-test-password", 32)...)
body = append(body, 0x01)
request := buildFrameWithDeclaredLength(0x05, 0xfe, "\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00", body, 41)
env, err := ParseFrame(request, 1782914969584, "8.134.95.166:39376")
if err != nil {
t.Fatal(err)
}
response, ok, err := AutoResponse(request, env)
if err != nil {
t.Fatalf("AutoResponse() error = %v", err)
}
if !ok {
t.Fatal("expected response")
}
if response[2] != 0x05 || response[3] != 0x01 {
t.Fatalf("unexpected response header: %x", response[:4])
}
if hex.EncodeToString(response[4:21]) != "0000000000000000000000000000000000" {
t.Fatalf("raw vin not echoed: %x", response[4:21])
}
if hex.EncodeToString(response[24:30]) != "1a070200260f" {
t.Fatalf("timestamp body not preserved: %x", response[24:30])
}
if got, want := bcc(response[2:len(response)-1]), response[len(response)-1]; got != want {
t.Fatalf("response bcc got=0x%02x want=0x%02x", got, want)
}
}
func TestParseFrameExtractsRealtimeVehicleMileageAndPosition(t *testing.T) {
body := []byte{0x1a, 0x06, 0x1e, 0x16, 0x17, 0x39}
body = append(body, 0x01)

80
go/vehicle-gateway/internal/protocol/gb32960/response.go Normal file
View File

@@ -0,0 +1,80 @@
package gb32960
import (
"encoding/binary"
"errors"
"time"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)
var ErrResponseFrameTooShort = errors.New("gb32960 response frame too short")
const (
responseSuccess = byte(0x01)
responseCommand = byte(0xfe)
encryptNone = byte(0x01)
)
// AutoResponse builds the protocol ACK for accepted upstream GB/T 32960 frames.
// The ACK is written only after the caller has durably published the frame.
func AutoResponse(raw []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
if env.ParseStatus == envelope.ParseBadFrame {
return nil, false, nil
}
if len(raw) < headerLen+1 {
return nil, false, ErrResponseFrameTooShort
}
if raw[3] != responseCommand {
return nil, false, nil
}
command := raw[2]
if !shouldRespond(command) {
return nil, false, nil
}
body := []byte(nil)
if timestamp := responseTime(raw, command); !timestamp.IsZero() {
body = encodeGBTime(timestamp)
}
return buildResponse(raw[0], command, responseSuccess, raw[4:21], body), true, nil
}
func shouldRespond(command byte) bool {
switch command {
case 0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08:
return true
default:
return false
}
}
func responseTime(raw []byte, command byte) time.Time {
bodyLen := len(raw) - headerLen - 1
if bodyLen >= 6 {
return parseGBTime(raw[headerLen : headerLen+6])
}
return time.Now().In(time.FixedZone("Asia/Shanghai", 8*3600))
}
func encodeGBTime(t time.Time) []byte {
local := t.In(time.FixedZone("Asia/Shanghai", 8*3600))
return []byte{
byte(local.Year() - 2000),
byte(local.Month()),
byte(local.Day()),
byte(local.Hour()),
byte(local.Minute()),
byte(local.Second()),
}
}
func buildResponse(start byte, command byte, responseFlag byte, vin17 []byte, body []byte) []byte {
out := make([]byte, 0, headerLen+len(body)+1)
out = append(out, start, start, command, responseFlag)
out = append(out, vin17[:17]...)
out = append(out, encryptNone, 0, 0)
binary.BigEndian.PutUint16(out[22:24], uint16(len(body)))
out = append(out, body...)
return append(out, bcc(out[2:]))
}

80
go/vehicle-gateway/internal/protocol/jt808/parser_test.go
View File

@@ -332,6 +332,86 @@ func TestParseFrameParsesAuthenticationToken(t *testing.T) {
}
}
func TestAutoResponderBuildsRegisterAck(t *testing.T) {
body := make([]byte, 0, 46)
body = binary.BigEndian.AppendUint16(body, 16)
body = binary.BigEndian.AppendUint16(body, 32)
body = appendFixedASCII(body, "YUTNG", 5)
body = appendFixedASCII(body, "ZK6105CHEVNPG4", 20)
body = appendFixedASCII(body, "DEV0001", 7)
body = append(body, 2)
body = append(body, []byte("TEST123")...)
request := buildFrame(0x0100, "013079963379", 9, body)
env, err := ParseFrame(request, 1782918600000, "115.231.168.135:43625")
if err != nil {
t.Fatal(err)
}
response, ok, err := NewAutoResponder("g7gps").Respond(request, env)
if err != nil {
t.Fatalf("Respond() error = %v", err)
}
if !ok {
t.Fatal("expected response")
}
frames, remainder, err := ExtractFrames(response)
if err != nil {
t.Fatalf("ExtractFrames() error = %v", err)
}
if len(frames) != 1 || len(remainder) != 0 {
t.Fatalf("unexpected response split frames=%d remainder=%s", len(frames), hex.EncodeToString(remainder))
}
payload := frames[0]
if binary.BigEndian.Uint16(payload[0:2]) != 0x8100 {
t.Fatalf("message id = %04x", binary.BigEndian.Uint16(payload[0:2]))
}
if hex.EncodeToString(payload[4:10]) != "013079963379" {
t.Fatalf("phone bcd = %x", payload[4:10])
}
if binary.BigEndian.Uint16(payload[12:14]) != 9 || payload[14] != 0 || string(payload[15:20]) != "g7gps" {
t.Fatalf("unexpected register ack body: %x", payload[12:len(payload)-1])
}
}
func TestAutoResponderBuildsVersionedGeneralAck(t *testing.T) {
request := buildVersionedFrame(0x0200, "14894135060", 12, make([]byte, 28))
env, err := ParseFrame(request, 1782918600000, "115.231.168.135:43625")
if err != nil {
t.Fatal(err)
}
response, ok, err := NewAutoResponder("").Respond(request, env)
if err != nil {
t.Fatalf("Respond() error = %v", err)
}
if !ok {
t.Fatal("expected response")
}
frames, remainder, err := ExtractFrames(response)
if err != nil {
t.Fatalf("ExtractFrames() error = %v", err)
}
if len(frames) != 1 || len(remainder) != 0 {
t.Fatalf("unexpected response split frames=%d remainder=%s", len(frames), hex.EncodeToString(remainder))
}
payload := frames[0]
if binary.BigEndian.Uint16(payload[0:2]) != 0x8001 {
t.Fatalf("message id = %04x", binary.BigEndian.Uint16(payload[0:2]))
}
if binary.BigEndian.Uint16(payload[2:4])&0x4000 == 0 || payload[4] != 1 {
t.Fatalf("versioned header not preserved: %x", payload[:5])
}
if hex.EncodeToString(payload[5:15]) != "00000000014894135060" {
t.Fatalf("phone bcd = %x", payload[5:15])
}
bodyStart := 17
if binary.BigEndian.Uint16(payload[bodyStart:bodyStart+2]) != 12 ||
binary.BigEndian.Uint16(payload[bodyStart+2:bodyStart+4]) != 0x0200 ||
payload[bodyStart+4] != 0 {
t.Fatalf("unexpected general ack body: %x", payload[bodyStart:len(payload)-1])
}
}
func TestExtractFramesHandlesEscapedPayload(t *testing.T) {
payload := []byte{0x02, 0x00, 0x00, 0x02, 0x01, 0x33, 0x07, 0x79, 0x54, 0x25, 0x00, 0x01, 0x7e, 0x7d}
frame := append([]byte{0x7e}, escape(append(payload, checksum(payload)))...)

130
go/vehicle-gateway/internal/protocol/jt808/response.go Normal file
View File

@@ -0,0 +1,130 @@
package jt808
import (
"encoding/binary"
"errors"
"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)
var ErrResponseHeaderTooShort = errors.New("jt808 response header too short")
const (
msgTerminalGeneralResponse = uint16(0x0001)
msgTerminalRegister = uint16(0x0100)
msgPlatformGeneralResponse = uint16(0x8001)
msgPlatformRegisterAck = uint16(0x8100)
)
type AutoResponder struct {
authCode string
}
func NewAutoResponder(authCode string) AutoResponder {
if authCode == "" {
authCode = "g7gps"
}
return AutoResponder{authCode: authCode}
}
func (r AutoResponder) Respond(raw []byte, env envelope.FrameEnvelope) ([]byte, bool, error) {
if env.ParseStatus == envelope.ParseBadFrame {
return nil, false, nil
}
header, err := parseResponseHeader(raw)
if err != nil {
return nil, false, err
}
if header.messageID == msgTerminalGeneralResponse {
return nil, false, nil
}
if header.messageID == msgTerminalRegister {
body := make([]byte, 3, 3+len(r.authCode))
binary.BigEndian.PutUint16(body[0:2], header.sequence)
body[2] = 0
body = append(body, []byte(r.authCode)...)
return encodeResponse(msgPlatformRegisterAck, header, body), true, nil
}
body := make([]byte, 5)
binary.BigEndian.PutUint16(body[0:2], header.sequence)
binary.BigEndian.PutUint16(body[2:4], header.messageID)
body[4] = 0
return encodeResponse(msgPlatformGeneralResponse, header, body), true, nil
}
type responseHeader struct {
messageID uint16
versioned bool
protocolVersion byte
phoneBCD []byte
sequence uint16
}
func parseResponseHeader(raw []byte) (responseHeader, error) {
if len(raw) < 12 {
return responseHeader{}, ErrResponseHeaderTooShort
}
props := binary.BigEndian.Uint16(raw[2:4])
versioned := props&0x4000 != 0
phoneStart := 4
phoneLen := 6
protocolVersion := byte(0)
if versioned {
if len(raw) < 17 {
return responseHeader{}, ErrResponseHeaderTooShort
}
protocolVersion = raw[4]
phoneStart = 5
phoneLen = 10
}
sequenceStart := phoneStart + phoneLen
if len(raw) < sequenceStart+2 {
return responseHeader{}, ErrResponseHeaderTooShort
}
phone := append([]byte(nil), raw[phoneStart:sequenceStart]...)
return responseHeader{
messageID: binary.BigEndian.Uint16(raw[0:2]),
versioned: versioned,
protocolVersion: protocolVersion,
phoneBCD: phone,
sequence: binary.BigEndian.Uint16(raw[sequenceStart : sequenceStart+2]),
}, nil
}
func encodeResponse(messageID uint16, header responseHeader, body []byte) []byte {
props := uint16(len(body)) & 0x03ff
if header.versioned {
props |= 0x4000
}
payload := make([]byte, 0, 4+1+len(header.phoneBCD)+2+len(body)+1)
payload = appendU16(payload, messageID)
payload = appendU16(payload, props)
if header.versioned {
payload = append(payload, header.protocolVersion)
}
payload = append(payload, header.phoneBCD...)
payload = appendU16(payload, header.sequence)
payload = append(payload, body...)
payload = append(payload, checksum(payload))
return frame(payload)
}
func appendU16(out []byte, value uint16) []byte {
return append(out, byte(value>>8), byte(value))
}
func frame(payload []byte) []byte {
out := []byte{0x7e}
for _, value := range payload {
switch value {
case 0x7e:
out = append(out, 0x7d, 0x02)
case 0x7d:
out = append(out, 0x7d, 0x01)
default:
out = append(out, value)
}
}
out = append(out, 0x7e)
return out
}