diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index f9a9dd3d..c59c2479 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -46,12 +46,14 @@ func main() { 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, }, } diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 18f52484..30e6fba9 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -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 { diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index 291e72b6..82d59dab 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -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{ diff --git a/go/vehicle-gateway/internal/protocol/gb32960/parser_test.go b/go/vehicle-gateway/internal/protocol/gb32960/parser_test.go index d172df72..cc7dc6f2 100644 --- a/go/vehicle-gateway/internal/protocol/gb32960/parser_test.go +++ b/go/vehicle-gateway/internal/protocol/gb32960/parser_test.go @@ -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) diff --git a/go/vehicle-gateway/internal/protocol/gb32960/response.go b/go/vehicle-gateway/internal/protocol/gb32960/response.go new file mode 100644 index 00000000..cb320ca8 --- /dev/null +++ b/go/vehicle-gateway/internal/protocol/gb32960/response.go @@ -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:])) +} diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go index 63f5ffff..6e0152f2 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go @@ -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)))...) diff --git a/go/vehicle-gateway/internal/protocol/jt808/response.go b/go/vehicle-gateway/internal/protocol/jt808/response.go new file mode 100644 index 00000000..b1f5a69b --- /dev/null +++ b/go/vehicle-gateway/internal/protocol/jt808/response.go @@ -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 +}