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

perf(go): gzip history raw field responses

Browse Source
This commit is contained in:
lingniu
2026-07-03 14:38:48 +08:00
parent 6318c10c41
commit 6624930cc1
2 changed files with 54 additions and 29 deletions
Download Patch File Download Diff File Expand all files Collapse all files

77
go/vehicle-gateway/internal/history/query.go
View File

@@ -1,6 +1,7 @@
package history
import (
"compress/gzip"
"context"
"database/sql"
"encoding/json"
@@ -34,25 +35,25 @@ type RawFrameQuery struct {
}
type RawFrameRow struct {
TS string `json:"ts"`
FrameID string `json:"frame_id"`
EventID string `json:"event_id"`
MessageID int64 `json:"message_id"`
MessageIDHex string `json:"message_id_hex"`
EventTime string `json:"event_time"`
ReceivedAt string `json:"received_at"`
RawSizeBytes int64 `json:"raw_size_bytes"`
RawHex string `json:"raw_hex,omitempty"`
RawText string `json:"raw_text,omitempty"`
ParsedFields string `json:"parsed_fields,omitempty"`
ParseStatus string `json:"parse_status"`
ParseError string `json:"parse_error,omitempty"`
SourceEndpoint string `json:"source_endpoint"`
Protocol string `json:"protocol"`
VehicleKey string `json:"vehicle_key"`
VIN string `json:"vin"`
Phone string `json:"phone,omitempty"`
DeviceID string `json:"device_id,omitempty"`
TS string `json:"ts"`
FrameID string `json:"frame_id"`
EventID string `json:"event_id"`
MessageID int64 `json:"message_id"`
MessageIDHex string `json:"message_id_hex"`
EventTime string `json:"event_time"`
ReceivedAt string `json:"received_at"`
RawSizeBytes int64 `json:"raw_size_bytes"`
RawHex string `json:"raw_hex,omitempty"`
RawText string `json:"raw_text,omitempty"`
ParsedFields json.RawMessage `json:"parsed_fields,omitempty"`
ParseStatus string `json:"parse_status"`
ParseError string `json:"parse_error,omitempty"`
SourceEndpoint string `json:"source_endpoint"`
Protocol string `json:"protocol"`
VehicleKey string `json:"vehicle_key"`
VIN string `json:"vin"`
Phone string `json:"phone,omitempty"`
DeviceID string `json:"device_id,omitempty"`
}
type LocationQuery struct {
@@ -128,6 +129,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
var ts scanDateTime
var eventTime scanDateTime
var receivedAt scanDateTime
var parsedFields string
if err := rows.Scan(
&ts,
&row.FrameID,
@@ -138,7 +140,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
&row.RawSizeBytes,
&row.RawHex,
&row.RawText,
&row.ParsedFields,
&parsedFields,
&row.ParseStatus,
&row.ParseError,
&row.SourceEndpoint,
@@ -154,6 +156,7 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
row.EventTime = eventTime.String
row.ReceivedAt = receivedAt.String
row.MessageIDHex = fmt.Sprintf("0x%04X", row.MessageID)
row.ParsedFields = rawJSONMessage(parsedFields)
out = append(out, row)
}
if err := rows.Err(); err != nil {
@@ -365,7 +368,7 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra
}{
{kind: "raw_hex", value: rows[index].RawHex},
{kind: "raw_text", value: rows[index].RawText},
{kind: "parsed_fields", value: rows[index].ParsedFields},
{kind: "parsed_fields", value: string(rows[index].ParsedFields)},
} {
manifest, ok := parsePayloadChunkManifest(candidate.value)
if !ok || !payloadKindMatches(candidate.kind, manifest.PayloadKind) || manifest.ChunkCount <= 0 {
@@ -437,12 +440,23 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra
case "raw_text":
rows[target.rowIndex].RawText = hydrated
case "parsed_fields":
rows[target.rowIndex].ParsedFields = hydrated
rows[target.rowIndex].ParsedFields = rawJSONMessage(hydrated)
}
}
return nil
}
func rawJSONMessage(value string) json.RawMessage {
value = strings.TrimSpace(value)
if value == "" {
return nil
}
if json.Valid([]byte(value)) {
return json.RawMessage(value)
}
return json.RawMessage(jsonString(value))
}
func payloadKindMatches(candidate string, manifest string) bool {
if candidate == manifest {
return true
@@ -610,8 +624,7 @@ func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if !query.IncludeTotal {
total = int64(len(rows))
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
writeHistoryJSON(w, r, map[string]any{
"items": rows,
"total": total,
"limit": query.Limit,
@@ -649,8 +662,7 @@ func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if !query.IncludeTotal {
total = int64(len(rows))
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
writeHistoryJSON(w, r, map[string]any{
"items": rows,
"total": total,
"limit": query.Limit,
@@ -658,6 +670,19 @@ func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
})
}
func writeHistoryJSON(w http.ResponseWriter, r *http.Request, value any) {
w.Header().Set("Content-Type", "application/json")
w.Header().Add("Vary", "Accept-Encoding")
if strings.Contains(strings.ToLower(r.Header.Get("Accept-Encoding")), "gzip") {
w.Header().Set("Content-Encoding", "gzip")
gzipWriter := gzip.NewWriter(w)
defer gzipWriter.Close()
_ = json.NewEncoder(gzipWriter).Encode(value)
return
}
_ = json.NewEncoder(w).Encode(value)
}
func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) {
values := r.URL.Query()
limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit")