mirror of
https://github.com/AmanTahiliani/box-box.git
synced 2026-08-07 19:56:18 -04:00
Implement SignalR Core support and add session fetching for live timing
This commit is contained in:
@@ -1,4 +1,4 @@
|
||||
import type { ArticleContent, LiveStateResponse, Meeting, NewsItem, RaceHub, Weekend } from './types'
|
||||
import type { ArticleContent, LiveStateResponse, Meeting, NewsItem, RaceHub, Session, Weekend } from './types'
|
||||
|
||||
export async function fetchRaceHub(sessionKey: number): Promise<RaceHub> {
|
||||
const res = await fetch(`/api/v1/race-hub?session_key=${sessionKey}`)
|
||||
@@ -35,6 +35,15 @@ export async function fetchSeasonMeetings(year: number): Promise<Meeting[]> {
|
||||
return Array.isArray(meetings) ? meetings : []
|
||||
}
|
||||
|
||||
export async function fetchSessions(meetingKey: number, source = 'openf1'): Promise<Session[]> {
|
||||
const res = await fetch(`/api/v1/sessions?meeting_key=${meetingKey}&source=${source}`)
|
||||
if (!res.ok) {
|
||||
throw new Error(`API ${res.status}: ${res.statusText}`)
|
||||
}
|
||||
const sessions = await res.json()
|
||||
return Array.isArray(sessions) ? sessions : []
|
||||
}
|
||||
|
||||
export async function fetchWeekend(meetingKey: number): Promise<Weekend> {
|
||||
const res = await fetch(`/api/v1/weekend?meeting_key=${meetingKey}`)
|
||||
if (!res.ok) {
|
||||
|
||||
@@ -1,8 +1,8 @@
|
||||
import { useEffect, useMemo, useState } from 'react'
|
||||
import { useQueries, useQuery } from '@tanstack/react-query'
|
||||
import { Link } from '@tanstack/react-router'
|
||||
import { fetchLiveState, fetchLocalMeetings, fetchSeasonMeetings, fetchSeasons, fetchWeekend } from '../api'
|
||||
import { countWeekendStats, formatCoverageHint, sessionTypeAbbrev } from '../lib/coverage'
|
||||
import { fetchLiveState, fetchLocalMeetings, fetchSeasonMeetings, fetchSeasons, fetchSessions, fetchWeekend } from '../api'
|
||||
import { RACE_HUB_DATASETS, countWeekendStats, formatCoverageHint, sessionTypeAbbrev } from '../lib/coverage'
|
||||
import {
|
||||
currentAndNextSession,
|
||||
focusMeetingKind,
|
||||
@@ -20,6 +20,10 @@ import { PaddockBriefing } from '../components/PaddockBriefing'
|
||||
|
||||
type WeekendStatusKind = 'live' | 'current' | 'next' | 'recent' | 'fallback'
|
||||
|
||||
const missingDatasets = Object.fromEntries(
|
||||
RACE_HUB_DATASETS.map((dataset) => [dataset, { status: 'missing', source: 'none', count: 0 }]),
|
||||
) as WeekendSession['datasets']
|
||||
|
||||
function classifySessionStatus(session: Session, now: Date): 'live' | 'done' | 'upcoming' {
|
||||
const start = sessionStartTime(session)
|
||||
const end = sessionEndTime(session)
|
||||
@@ -70,13 +74,13 @@ export function CommandCenterPage() {
|
||||
|
||||
const localMeetings = meetingsQuery.data ?? []
|
||||
const seasonMeetings = seasonMeetingsQuery.data?.length ? seasonMeetingsQuery.data : localMeetings
|
||||
const meetings = localMeetings
|
||||
const focusMeetings = seasonMeetings.length > 0 ? seasonMeetings : localMeetings
|
||||
|
||||
const weekendQueries = useQueries({
|
||||
queries: meetings.map((meeting) => ({
|
||||
queries: localMeetings.map((meeting) => ({
|
||||
queryKey: ['weekend', meeting.meeting_key],
|
||||
queryFn: () => fetchWeekend(meeting.meeting_key),
|
||||
enabled: meetings.length > 0,
|
||||
enabled: localMeetings.length > 0,
|
||||
staleTime: 60_000,
|
||||
})),
|
||||
})
|
||||
@@ -96,28 +100,45 @@ export function CommandCenterPage() {
|
||||
|
||||
const weekendsByKey = useMemo(() => {
|
||||
const map = new Map<number, Weekend>()
|
||||
meetings.forEach((meeting, i) => {
|
||||
localMeetings.forEach((meeting, i) => {
|
||||
const data = weekendQueries[i]?.data
|
||||
if (data) map.set(meeting.meeting_key, data)
|
||||
})
|
||||
return map
|
||||
}, [meetings, weekendQueries])
|
||||
}, [localMeetings, weekendQueries])
|
||||
|
||||
const weekendList = useMemo(() => weekendQueries.map((q) => q.data), [weekendQueries])
|
||||
const meetingStats = countWeekendStats(weekendList)
|
||||
const focusMeeting = pickFocusMeeting(meetings, nowDate)
|
||||
const focusMeeting = pickFocusMeeting(focusMeetings, nowDate)
|
||||
const focusWeekend = focusMeeting ? weekendsByKey.get(focusMeeting.meeting_key) : undefined
|
||||
const openF1SessionsQuery = useQuery({
|
||||
queryKey: ['sessions', focusMeeting?.meeting_key, 'openf1'],
|
||||
queryFn: () => fetchSessions(focusMeeting!.meeting_key, 'openf1'),
|
||||
enabled: focusMeeting != null && focusWeekend == null,
|
||||
staleTime: 60_000,
|
||||
})
|
||||
const openF1WeekendSessions: WeekendSession[] = useMemo(
|
||||
() =>
|
||||
(openF1SessionsQuery.data ?? []).map((session) => ({
|
||||
session,
|
||||
source: 'none',
|
||||
datasets: missingDatasets,
|
||||
})),
|
||||
[openF1SessionsQuery.data],
|
||||
)
|
||||
const focusWeekendSessions = focusWeekend?.sessions ?? openF1WeekendSessions
|
||||
const focusKind = focusMeeting ? focusMeetingKind(focusMeeting, nowDate) : null
|
||||
const focusSessions: Session[] = focusWeekend
|
||||
? sortSessionsByStart(focusWeekend.sessions.map((s) => s.session))
|
||||
: []
|
||||
const focusSessions: Session[] = sortSessionsByStart(focusWeekendSessions.map((s) => s.session))
|
||||
const { current: currentSession, next: nextSession } = currentAndNextSession(focusSessions, nowDate)
|
||||
|
||||
const analysisSession = pickAnalysisSession(focusWeekend)
|
||||
const analysisSessionKey =
|
||||
analysisSession?.session.session_key ??
|
||||
focusWeekend?.default_session_key ??
|
||||
focusWeekend?.sessions[0]?.session.session_key
|
||||
const actionSession =
|
||||
analysisSession?.session ??
|
||||
currentSession ??
|
||||
nextSession ??
|
||||
focusWeekendSessions.find((s) => s.session.session_key === focusWeekend?.default_session_key)?.session ??
|
||||
focusWeekendSessions[0]?.session
|
||||
const analysisSessionKey = actionSession?.session_key
|
||||
|
||||
const weekendsLoading = weekendQueries.some((q) => q.isLoading)
|
||||
|
||||
@@ -243,8 +264,8 @@ export function CommandCenterPage() {
|
||||
>
|
||||
<span className="cc-pri-label">Open Analysis</span>
|
||||
<span className="cc-pri-meta mono">
|
||||
{analysisSession
|
||||
? `${analysisSession.session.session_name} · session ${analysisSession.session.session_key}`
|
||||
{actionSession
|
||||
? `${actionSession.session_name} · session ${actionSession.session_key}`
|
||||
: 'Pick a session'}
|
||||
</span>
|
||||
</Link>
|
||||
@@ -267,14 +288,14 @@ export function CommandCenterPage() {
|
||||
</div>
|
||||
)}
|
||||
|
||||
{focusWeekend && focusWeekend.sessions.length > 0 && (
|
||||
{focusWeekendSessions.length > 0 && (
|
||||
<section className="cc-schedule" id="cc-schedule">
|
||||
<div className="sec-header">
|
||||
<span className="sec-title">Weekend Schedule</span>
|
||||
<span className="sec-meta mono">{focusWeekend.sessions.length} sessions</span>
|
||||
<span className="sec-meta mono">{focusWeekendSessions.length} sessions</span>
|
||||
</div>
|
||||
<div className="cc-session-strip" role="list">
|
||||
{focusWeekend.sessions.map(({ session, source, datasets }) => {
|
||||
{focusWeekendSessions.map(({ session, source, datasets }) => {
|
||||
const status = classifySessionStatus(session, nowDate)
|
||||
const isNext = nextSession?.session_key === session.session_key
|
||||
const isCurrent = currentSession?.session_key === session.session_key
|
||||
|
||||
@@ -9,15 +9,17 @@ vi.mock('../api', () => ({
|
||||
fetchSeasons: vi.fn(),
|
||||
fetchLocalMeetings: vi.fn(),
|
||||
fetchSeasonMeetings: vi.fn(),
|
||||
fetchSessions: vi.fn(),
|
||||
fetchWeekend: vi.fn(),
|
||||
fetchLiveState: vi.fn(),
|
||||
}))
|
||||
|
||||
import { fetchSeasons, fetchLocalMeetings, fetchSeasonMeetings, fetchWeekend, fetchLiveState } from '../api'
|
||||
import { fetchSeasons, fetchLocalMeetings, fetchSeasonMeetings, fetchSessions, fetchWeekend, fetchLiveState } from '../api'
|
||||
|
||||
const mockFetchSeasons = vi.mocked(fetchSeasons)
|
||||
const mockFetchLocalMeetings = vi.mocked(fetchLocalMeetings)
|
||||
const mockFetchSeasonMeetings = vi.mocked(fetchSeasonMeetings)
|
||||
const mockFetchSessions = vi.mocked(fetchSessions)
|
||||
const mockFetchWeekend = vi.mocked(fetchWeekend)
|
||||
const mockFetchLiveState = vi.mocked(fetchLiveState)
|
||||
|
||||
@@ -99,6 +101,7 @@ describe('CommandCenterPage', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks()
|
||||
mockFetchLiveState.mockResolvedValue({ is_live: false, data: null })
|
||||
mockFetchSessions.mockResolvedValue([])
|
||||
})
|
||||
|
||||
it('shows empty state when no seasons are ingested', async () => {
|
||||
@@ -134,4 +137,54 @@ describe('CommandCenterPage', () => {
|
||||
expect(screen.getByText('No live session')).toBeInTheDocument()
|
||||
expect(screen.getByTestId('cc-action-race-hub')).toHaveTextContent('Race')
|
||||
})
|
||||
|
||||
it('uses OpenF1 calendar metadata to focus the current weekend when local ingest is behind', async () => {
|
||||
const canada = {
|
||||
...meeting,
|
||||
meeting_key: 1301,
|
||||
meeting_name: 'Canada',
|
||||
country_name: 'Canada',
|
||||
country_code: 'CAN',
|
||||
circuit_short_name: 'Montreal',
|
||||
date_start: '2026-05-22T00:00:00+00:00',
|
||||
date_end: '2026-05-24T23:59:59+00:00',
|
||||
year: 2026,
|
||||
}
|
||||
const monaco = {
|
||||
...meeting,
|
||||
meeting_key: 1302,
|
||||
meeting_name: 'Monaco',
|
||||
date_start: '2026-06-05T00:00:00+00:00',
|
||||
date_end: '2026-06-07T23:59:59+00:00',
|
||||
year: 2026,
|
||||
}
|
||||
|
||||
vi.setSystemTime(new Date('2026-06-06T15:15:00Z'))
|
||||
mockFetchSeasons.mockResolvedValue([2026])
|
||||
mockFetchLocalMeetings.mockResolvedValue([canada])
|
||||
mockFetchSeasonMeetings.mockResolvedValue([canada, monaco])
|
||||
mockFetchWeekend.mockResolvedValue({ ...weekend, meeting: canada, meeting_key: canada.meeting_key })
|
||||
mockFetchSessions.mockResolvedValue([
|
||||
{
|
||||
session_key: 9602,
|
||||
session_name: 'Qualifying',
|
||||
session_type: 'Qualifying',
|
||||
meeting_key: monaco.meeting_key,
|
||||
date_start: '2026-06-06T15:00:00+00:00',
|
||||
date_end: '2026-06-06T16:00:00+00:00',
|
||||
gmt_offset: '02:00:00',
|
||||
},
|
||||
])
|
||||
|
||||
renderPage()
|
||||
|
||||
await waitFor(() => {
|
||||
expect(screen.getByTestId('cc-focus')).toHaveTextContent('Monaco')
|
||||
})
|
||||
expect(screen.getByTestId('cc-focus')).toHaveTextContent('Current weekend')
|
||||
expect(screen.getByTestId('cc-session-9602')).toHaveTextContent('On track')
|
||||
expect(screen.getByTestId('cc-action-race-hub')).toHaveTextContent('Qualifying')
|
||||
|
||||
vi.useRealTimers()
|
||||
})
|
||||
})
|
||||
|
||||
@@ -77,6 +77,41 @@ func TestProcessMessageIncremental(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessCoreMessageCompletionAndFeed(t *testing.T) {
|
||||
state := live.NewState()
|
||||
msg := []byte(`{"type":3,"invocationId":"1","result":{
|
||||
"ExtrapolatedClock":{"Remaining":"00:04:23","Utc":"2026-06-06T14:42:36.0491737Z","Extrapolating":true},
|
||||
"TimingData":{"Lines":{"12":{"Position":"1","RacingNumber":"12","Sectors":[{"Value":"22.430"},{"Value":"40.119"},{"Value":""}],"Speeds":{"ST":{"Value":"254"}},"BestLapTime":{"Value":"1:12.704","Lap":15}}}},
|
||||
"DriverList":{"12":{"RacingNumber":"12","Tla":"ANT","TeamName":"Mercedes","TeamColour":"00D7B6"}}
|
||||
}}` + "\x1e")
|
||||
|
||||
if !state.ProcessCoreMessage(msg) {
|
||||
t.Fatal("expected SignalR Core completion to produce updates")
|
||||
}
|
||||
|
||||
snap := state.Snapshot()
|
||||
if snap.Clock != "00:04:23" || !snap.ClockExtrapolating {
|
||||
t.Errorf("clock = %q extrapolating=%v", snap.Clock, snap.ClockExtrapolating)
|
||||
}
|
||||
if snap.Drivers["12"].Position != 1 || snap.Drivers["12"].BestLapTime != "1:12.704" {
|
||||
t.Errorf("driver = %+v", snap.Drivers["12"])
|
||||
}
|
||||
if snap.Drivers["12"].Sectors[1].Value != "40.119" || snap.Drivers["12"].SpeedTrap != "254" {
|
||||
t.Errorf("driver sectors/speed = %+v", snap.Drivers["12"])
|
||||
}
|
||||
if snap.DriverInfo["12"].Tla != "ANT" {
|
||||
t.Errorf("driver info = %+v", snap.DriverInfo["12"])
|
||||
}
|
||||
|
||||
feed := []byte(`{"type":1,"target":"feed","arguments":["TimingData",{"Lines":{"12":{"LastLapTime":{"Value":"1:13.000","PersonalFastest":true}}}},"2026-06-06T14:42:37Z"]}` + "\x1e")
|
||||
if !state.ProcessCoreMessage(feed) {
|
||||
t.Fatal("expected SignalR Core feed frame to produce updates")
|
||||
}
|
||||
if state.Snapshot().Drivers["12"].LastLapTime != "1:13.000" {
|
||||
t.Errorf("last lap = %q", state.Snapshot().Drivers["12"].LastLapTime)
|
||||
}
|
||||
}
|
||||
|
||||
func TestProcessTopicTimingData(t *testing.T) {
|
||||
state := live.NewState()
|
||||
data := json.RawMessage(`{
|
||||
|
||||
@@ -1,11 +1,14 @@
|
||||
package live
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"os"
|
||||
|
||||
"github.com/gorilla/websocket"
|
||||
)
|
||||
@@ -14,6 +17,132 @@ import (
|
||||
// to timing topics, and sends defensive snapshots on dataChan until the
|
||||
// connection closes.
|
||||
func ConnectToF1LiveTiming(dataChan chan LiveStreamData) error {
|
||||
if err := connectToF1SignalRCore(dataChan); err == nil {
|
||||
return nil
|
||||
} else {
|
||||
log.Printf("f1 signalrcore feed unavailable, trying legacy signalr: %v", err)
|
||||
}
|
||||
return connectToF1LegacySignalR(dataChan)
|
||||
}
|
||||
|
||||
func connectToF1SignalRCore(dataChan chan LiveStreamData) error {
|
||||
req, err := http.NewRequest("POST", "https://livetiming.formula1.com/signalrcore/negotiate?negotiateVersion=1", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Origin", "https://www.formula1.com")
|
||||
req.Header.Set("User-Agent", "Mozilla/5.0")
|
||||
req.Header.Set("Content-Length", "0")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("f1 signalrcore negotiate returned %s", resp.Status)
|
||||
}
|
||||
|
||||
var neg struct {
|
||||
ConnectionToken string `json:"connectionToken"`
|
||||
}
|
||||
if err := json.Unmarshal(body, &neg); err != nil {
|
||||
return err
|
||||
}
|
||||
if neg.ConnectionToken == "" {
|
||||
return fmt.Errorf("f1 signalrcore negotiate returned an empty connection token")
|
||||
}
|
||||
|
||||
wsURL := "wss://livetiming.formula1.com/signalrcore?id=" + url.QueryEscape(neg.ConnectionToken)
|
||||
header := http.Header{}
|
||||
header.Set("Origin", "https://www.formula1.com")
|
||||
header.Set("User-Agent", "Mozilla/5.0")
|
||||
for _, cookie := range resp.Cookies() {
|
||||
header.Add("Cookie", cookie.String())
|
||||
}
|
||||
|
||||
c, _, err := websocket.DefaultDialer.Dial(wsURL, header)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
writeCoreFrame := func(payload any) error {
|
||||
body, err := json.Marshal(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
body = append(body, signalRRecordSeparator)
|
||||
return c.WriteMessage(websocket.TextMessage, body)
|
||||
}
|
||||
|
||||
if err := writeCoreFrame(map[string]any{"protocol": "json", "version": 1}); err != nil {
|
||||
c.Close()
|
||||
return err
|
||||
}
|
||||
_, handshake, err := c.ReadMessage()
|
||||
if err != nil {
|
||||
c.Close()
|
||||
return err
|
||||
}
|
||||
if !bytes.Contains(handshake, []byte("{}")) {
|
||||
c.Close()
|
||||
return fmt.Errorf("f1 signalrcore handshake returned %q", string(handshake))
|
||||
}
|
||||
|
||||
topics := []string{
|
||||
"Heartbeat",
|
||||
"TimingData",
|
||||
"DriverList",
|
||||
"LapCount",
|
||||
"ExtrapolatedClock",
|
||||
"TrackStatus",
|
||||
"RaceControlMessages",
|
||||
"WeatherData",
|
||||
"SessionInfo",
|
||||
"CurrentTyres",
|
||||
"TimingAppData",
|
||||
"TimingStats",
|
||||
"SessionStatus",
|
||||
"TopThree",
|
||||
}
|
||||
if err := writeCoreFrame(map[string]any{
|
||||
"type": 1,
|
||||
"target": "subscribe",
|
||||
"arguments": []any{topics},
|
||||
"invocationId": "1",
|
||||
}); err != nil {
|
||||
c.Close()
|
||||
return err
|
||||
}
|
||||
|
||||
go func() {
|
||||
defer c.Close()
|
||||
state := NewState()
|
||||
|
||||
for {
|
||||
_, message, err := c.ReadMessage()
|
||||
if err != nil {
|
||||
log.Println("SignalR Core read error:", err)
|
||||
return
|
||||
}
|
||||
|
||||
if state.ProcessCoreMessage(message) {
|
||||
select {
|
||||
case dataChan <- state.Snapshot():
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func connectToF1LegacySignalR(dataChan chan LiveStreamData) error {
|
||||
hubName := `[{"name":"Streaming"}]`
|
||||
negotiateURL := fmt.Sprintf("https://livetiming.formula1.com/signalr/negotiate?clientProtocol=1.5&connectionData=%s", url.QueryEscape(hubName))
|
||||
|
||||
@@ -21,20 +150,30 @@ func ConnectToF1LiveTiming(dataChan chan LiveStreamData) error {
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("User-Agent", "BestHTTP")
|
||||
if token := f1LiveBearerToken(); token != "" {
|
||||
req.Header.Set("Authorization", "Bearer "+token)
|
||||
}
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
cookies := resp.Cookies()
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("f1 signalr negotiate returned %s (auth=%s)", resp.Status, authState())
|
||||
}
|
||||
|
||||
cookies := resp.Cookies()
|
||||
var neg struct {
|
||||
ConnectionToken string `json:"ConnectionToken"`
|
||||
}
|
||||
if err := json.NewDecoder(resp.Body).Decode(&neg); err != nil {
|
||||
return err
|
||||
}
|
||||
if neg.ConnectionToken == "" {
|
||||
return fmt.Errorf("f1 signalr negotiate returned an empty connection token")
|
||||
}
|
||||
|
||||
wsURL := fmt.Sprintf("wss://livetiming.formula1.com/signalr/connect?clientProtocol=1.5&transport=webSockets&connectionToken=%s&connectionData=%s",
|
||||
url.QueryEscape(neg.ConnectionToken),
|
||||
@@ -46,6 +185,9 @@ func ConnectToF1LiveTiming(dataChan chan LiveStreamData) error {
|
||||
header.Add("Cookie", cookie.String())
|
||||
}
|
||||
header.Add("User-Agent", "BestHTTP")
|
||||
if token := f1LiveBearerToken(); token != "" {
|
||||
header.Add("Authorization", "Bearer "+token)
|
||||
}
|
||||
|
||||
c, _, err := websocket.DefaultDialer.Dial(wsURL, header)
|
||||
if err != nil {
|
||||
@@ -80,3 +222,17 @@ func ConnectToF1LiveTiming(dataChan chan LiveStreamData) error {
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func f1LiveBearerToken() string {
|
||||
if token := os.Getenv("BOXBOX_F1_LIVE_BEARER_TOKEN"); token != "" {
|
||||
return token
|
||||
}
|
||||
return os.Getenv("F1_LIVE_BEARER_TOKEN")
|
||||
}
|
||||
|
||||
func authState() string {
|
||||
if f1LiveBearerToken() != "" {
|
||||
return "bearer-configured"
|
||||
}
|
||||
return "no-bearer-token"
|
||||
}
|
||||
|
||||
@@ -23,6 +23,8 @@ type State struct {
|
||||
ClockExtrapolating bool
|
||||
}
|
||||
|
||||
const signalRRecordSeparator = byte(0x1e)
|
||||
|
||||
// NewState returns an empty live timing accumulator.
|
||||
func NewState() *State {
|
||||
return &State{
|
||||
@@ -106,6 +108,51 @@ func (s *State) ProcessMessage(message []byte) bool {
|
||||
return updated
|
||||
}
|
||||
|
||||
// ProcessCoreMessage parses one or more SignalR Core JSON frames and applies
|
||||
// completion snapshots and feed deltas from the current official F1 live timing hub.
|
||||
func (s *State) ProcessCoreMessage(message []byte) bool {
|
||||
updated := false
|
||||
for _, frame := range splitSignalRFrames(message) {
|
||||
var envelope struct {
|
||||
Type int `json:"type"`
|
||||
Target string `json:"target"`
|
||||
Args []json.RawMessage `json:"arguments"`
|
||||
Result json.RawMessage `json:"result"`
|
||||
}
|
||||
if err := json.Unmarshal(frame, &envelope); err != nil {
|
||||
continue
|
||||
}
|
||||
|
||||
switch envelope.Type {
|
||||
case 1:
|
||||
if envelope.Target != "feed" || len(envelope.Args) < 2 {
|
||||
continue
|
||||
}
|
||||
var topic string
|
||||
if err := json.Unmarshal(envelope.Args[0], &topic); err != nil {
|
||||
continue
|
||||
}
|
||||
if s.ProcessTopic(topic, envelope.Args[1]) {
|
||||
updated = true
|
||||
}
|
||||
case 3:
|
||||
if len(envelope.Result) == 0 || string(envelope.Result) == "null" {
|
||||
continue
|
||||
}
|
||||
var result map[string]json.RawMessage
|
||||
if err := json.Unmarshal(envelope.Result, &result); err != nil {
|
||||
continue
|
||||
}
|
||||
for topic, data := range result {
|
||||
if s.ProcessTopic(topic, data) {
|
||||
updated = true
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
return updated
|
||||
}
|
||||
|
||||
// ProcessTopic applies a single topic payload to the accumulator.
|
||||
func (s *State) ProcessTopic(topic string, data json.RawMessage) bool {
|
||||
updated := false
|
||||
@@ -181,10 +228,10 @@ func (s *State) ProcessTopic(topic string, data json.RawMessage) bool {
|
||||
}
|
||||
case "RaceControlMessages":
|
||||
var rcm struct {
|
||||
Messages map[string]json.RawMessage `json:"Messages"`
|
||||
Messages json.RawMessage `json:"Messages"`
|
||||
}
|
||||
if json.Unmarshal(data, &rcm) == nil {
|
||||
for _, msgRaw := range rcm.Messages {
|
||||
for _, msgRaw := range indexedRawValues(rcm.Messages) {
|
||||
var msg struct {
|
||||
Utc string `json:"Utc"`
|
||||
Category string `json:"Category"`
|
||||
@@ -192,7 +239,7 @@ func (s *State) ProcessTopic(topic string, data json.RawMessage) bool {
|
||||
Message string `json:"Message"`
|
||||
Lap int `json:"Lap"`
|
||||
}
|
||||
if json.Unmarshal(msgRaw, &msg) == nil && msg.Message != "" {
|
||||
if json.Unmarshal(msgRaw.Raw, &msg) == nil && msg.Message != "" {
|
||||
t := ""
|
||||
if len(msg.Utc) >= 19 {
|
||||
t = msg.Utc[11:16]
|
||||
@@ -285,17 +332,17 @@ func (s *State) ProcessTopic(topic string, data json.RawMessage) bool {
|
||||
if json.Unmarshal(data, &tad) == nil {
|
||||
for num, lineRaw := range tad.Lines {
|
||||
var line struct {
|
||||
Stints map[string]json.RawMessage `json:"Stints"`
|
||||
Stints json.RawMessage `json:"Stints"`
|
||||
}
|
||||
if json.Unmarshal(lineRaw, &line) == nil && line.Stints != nil {
|
||||
var driverStints []LiveStintData
|
||||
for _, sRaw := range line.Stints {
|
||||
for _, sRaw := range indexedRawValues(line.Stints) {
|
||||
var st struct {
|
||||
Compound string `json:"Compound"`
|
||||
New string `json:"New"`
|
||||
TotalLaps int `json:"TotalLaps"`
|
||||
}
|
||||
if json.Unmarshal(sRaw, &st) == nil && st.Compound != "" {
|
||||
if json.Unmarshal(sRaw.Raw, &st) == nil && st.Compound != "" {
|
||||
driverStints = append(driverStints, LiveStintData{
|
||||
Compound: st.Compound,
|
||||
New: st.New == "true" || st.New == "True",
|
||||
@@ -417,16 +464,15 @@ func updateDriver(drivers map[string]LiveDriverData, num string, line F1TimingLi
|
||||
}
|
||||
}
|
||||
|
||||
for idx, sRaw := range line.Sectors {
|
||||
i := 0
|
||||
fmt.Sscanf(idx, "%d", &i)
|
||||
for _, sector := range indexedRawValues(line.Sectors) {
|
||||
i := sector.Index
|
||||
if i >= 0 && i < 3 {
|
||||
var sec struct {
|
||||
Value string `json:"Value"`
|
||||
PersonalFastest bool `json:"PersonalFastest"`
|
||||
OverallFastest bool `json:"OverallFastest"`
|
||||
}
|
||||
if json.Unmarshal(sRaw, &sec) == nil {
|
||||
if json.Unmarshal(sector.Raw, &sec) == nil {
|
||||
if sec.Value == "" {
|
||||
d.Sectors[i] = LiveSectorData{}
|
||||
} else {
|
||||
@@ -493,3 +539,54 @@ func toInt(v interface{}) (int, bool) {
|
||||
}
|
||||
return 0, false
|
||||
}
|
||||
|
||||
type indexedRaw struct {
|
||||
Index int
|
||||
Raw json.RawMessage
|
||||
}
|
||||
|
||||
func splitSignalRFrames(message []byte) []json.RawMessage {
|
||||
parts := []json.RawMessage{}
|
||||
start := 0
|
||||
for i, b := range message {
|
||||
if b != signalRRecordSeparator {
|
||||
continue
|
||||
}
|
||||
if i > start {
|
||||
parts = append(parts, json.RawMessage(message[start:i]))
|
||||
}
|
||||
start = i + 1
|
||||
}
|
||||
if start < len(message) {
|
||||
parts = append(parts, json.RawMessage(message[start:]))
|
||||
}
|
||||
return parts
|
||||
}
|
||||
|
||||
func indexedRawValues(raw json.RawMessage) []indexedRaw {
|
||||
if len(raw) == 0 || string(raw) == "null" {
|
||||
return nil
|
||||
}
|
||||
|
||||
var arr []json.RawMessage
|
||||
if err := json.Unmarshal(raw, &arr); err == nil {
|
||||
values := make([]indexedRaw, 0, len(arr))
|
||||
for i, v := range arr {
|
||||
values = append(values, indexedRaw{Index: i, Raw: v})
|
||||
}
|
||||
return values
|
||||
}
|
||||
|
||||
var obj map[string]json.RawMessage
|
||||
if err := json.Unmarshal(raw, &obj); err == nil {
|
||||
values := make([]indexedRaw, 0, len(obj))
|
||||
for k, v := range obj {
|
||||
i := 0
|
||||
fmt.Sscanf(k, "%d", &i)
|
||||
values = append(values, indexedRaw{Index: i, Raw: v})
|
||||
}
|
||||
return values
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
@@ -38,7 +38,7 @@ type F1TimingLine struct {
|
||||
KnockedOut interface{} `json:"KnockedOut"`
|
||||
Cutoff interface{} `json:"Cutoff"`
|
||||
NumberOfLaps interface{} `json:"NumberOfLaps"`
|
||||
Sectors map[string]json.RawMessage `json:"Sectors"`
|
||||
Sectors json.RawMessage `json:"Sectors"`
|
||||
Speeds map[string]json.RawMessage `json:"Speeds"`
|
||||
}
|
||||
|
||||
|
||||
@@ -121,8 +121,13 @@ func (s *Server) signalRLoop() {
|
||||
|
||||
s.hub.mu.Lock()
|
||||
s.hub.isLive = false
|
||||
s.hub.lastSnapshot = nil
|
||||
s.hub.mu.Unlock()
|
||||
|
||||
if payload, err := json.Marshal(map[string]any{"data": nil, "is_live": false}); err == nil {
|
||||
s.hub.broadcast <- sseEvent{name: "snapshot", data: payload}
|
||||
}
|
||||
|
||||
log.Printf("web: live feed reconnecting in %v", backoff)
|
||||
time.Sleep(backoff)
|
||||
if backoff < maxBackoff {
|
||||
|
||||
Reference in New Issue
Block a user