lplex

package module

Go to main page

Versions in this module

v0

v0.4.1

Mar 25, 2026

v0.4.0

Mar 24, 2026

Changes in this version

+ const DefaultClaimHeartbeat

+ const DefaultProductInfoHeartbeat

+ func ApplySlots(broker *Broker, slots []ClientSlot, logger *slog.Logger)

+ func CompressHandler(next http.Handler) http.Handler

+ func InitTracing(ctx context.Context, cfg TracingConfig) (shutdown func(context.Context) error, err error)

+ func LivenessHandler() http.HandlerFunc

+ func ReadinessHandler(cfg HealthConfig) http.HandlerFunc

+ func SDNotify(state string) bool

+ func SDReady() bool

+ func SDWatchdog() bool

+ func Tracer(name string) trace.Tracer

+ type AlertEvent struct

+ Details json.RawMessage

+ InstanceID string

+ Severity string

+ Summary string

+ Timestamp time.Time

+ Type AlertType

+ type AlertManager struct

+ func NewAlertManager(cfg AlertManagerConfig) *AlertManager

+ func (am *AlertManager) Fire(event AlertEvent)

+ func (am *AlertManager) FireBusResumed(silenceDuration time.Duration)

+ func (am *AlertManager) FireBusSilence(lastFrame time.Time, elapsed time.Duration)

+ func (am *AlertManager) FireDeviceRemoved(src uint8, dev *Device)

+ func (am *AlertManager) FireReplicationDisconnected(reason string)

+ func (am *AlertManager) FireReplicationReconnected()

+ func (am *AlertManager) Run(ctx context.Context)

+ type AlertManagerConfig struct

+ DedupWindow time.Duration

+ InstanceID string

+ Logger *slog.Logger

+ Timeout time.Duration

+ WebhookURL string

+ type AlertType string

+ const AlertBusResumed

+ const AlertBusSilence

+ const AlertDeviceRemoved

+ const AlertReplicationDisconnected

+ const AlertReplicationReconnected

type Broker

+ func (b *Broker) QueueTx(req TxRequest) bool

+ func (b *Broker) SetAlerts(am *AlertManager)

+ func (b *Broker) VirtualDevices() *VirtualDeviceManager

type BrokerConfig

+ ClaimHeartbeat time.Duration

+ DeviceIdleTimeout time.Duration

+ ProductInfoHeartbeat time.Duration

+ VirtualDevices []VirtualDeviceConfig

type BrokerStats

+ ConsumerMaxLag uint64

+ DevicesAdded uint64

+ DevicesRemoved uint64

+ JournalDrops uint64

+ type BusSource struct

+ Bus string

+ Source uint8

+ type CANBus interface

+ Name func() string

+ ReadFrames func(ctx context.Context, rxFrames chan<- RxFrame) error

+ WriteFrames func(ctx context.Context, txFrames <-chan TxRequest) error

+ type ClaimState int

+ const ClaimCannotClaim

+ const ClaimHeld

+ const ClaimIdle

+ const ClaimInProgress

+ func (s ClaimState) String() string

+ type ClientSlot struct

+ BufferTimeout time.Duration

+ Filter *EventFilter

+ ID string

+ func ParseClientSlot(cfg ClientSlotConfig) (ClientSlot, error)

+ type ClientSlotConfig struct

+ BufferTimeout string

+ Filter *SlotFilterConfig

+ ID string

type DecodedDeviceValues

+ Bus string

type Device

+ Bus string

+ type DeviceNotFoundError struct

+ Bus string

+ Source uint8

+ func (e *DeviceNotFoundError) Error() string

type DeviceRegistry

+ func (r *DeviceRegistry) ExpireIdle(cutoff time.Time) []BusSource

+ func (r *DeviceRegistry) SourcesMissingProductInfo() []BusSource

type DeviceValues

+ Bus string

type EventFilter

+ Buses []string

type Frame

+ Bus string

type InstanceManager

+ func (im *InstanceManager) SetDeviceIdleTimeout(d time.Duration)

+ func (im *InstanceManager) SetRingSize(size int)

type JournalWriter

+ func (w *JournalWriter) Stats() JournalWriterStats

+ type JournalWriterStats struct

+ BlocksWritten uint64

+ BytesWritten uint64

+ LastBlockWriteDuration time.Duration

+ type LoopbackBus struct

+ func NewLoopbackBus(name string, bufSize int, logger *slog.Logger) *LoopbackBus

+ func (b *LoopbackBus) Inject(frame RxFrame) bool

+ func (b *LoopbackBus) Name() string

+ func (b *LoopbackBus) ReadFrames(ctx context.Context, rxFrames chan<- RxFrame) error

+ func (b *LoopbackBus) WriteFrames(ctx context.Context, txFrames <-chan TxRequest) error

+ type MQTTBridge struct

+ func NewMQTTBridge(cfg MQTTBridgeConfig, broker *Broker) *MQTTBridge

+ func (m *MQTTBridge) Run(ctx context.Context) error

+ type MQTTBridgeConfig struct

+ BrokerURL string

+ ClientID string

+ Filter *EventFilter

+ Logger *slog.Logger

+ Password string

+ QoS byte

+ TopicPrefix string

+ Username string

type ReplicationClient

+ func (c *ReplicationClient) SetAlerts(am *AlertManager)

type ReplicationStatus

+ BackfillBlocksSent uint64

+ BackfillBytesSent uint64

+ LiveFramesSent uint64

+ Reconnects uint64

type RxFrame

+ Bus string

+ type SequenceGapError struct

+ ActualSeq uint64

+ ExpectedSeq uint64

+ func (e *SequenceGapError) Error() string

type Server

+ func (s *Server) HandleWebSocket(w http.ResponseWriter, r *http.Request)

+ func (s *Server) SetAPIKey(key string)

+ func (s *Server) SetReadOnly(readOnly bool)

+ func (s *Server) SetSendPolicy(policy sendpolicy.SendPolicy)

+ func (s *Server) SetSendRateLimit(rps float64, burst int)

+ type SessionNotFoundError struct

+ SessionID string

+ func (e *SessionNotFoundError) Error() string

+ type SlotFilterConfig struct

+ Bus []string

+ ExcludeName []string

+ ExcludePGN []uint32

+ Instance []uint8

+ Manufacturer []string

+ Name []string

+ PGN []uint32

+ type SocketCANBus struct

+ func NewSocketCANBus(iface string, logger *slog.Logger) *SocketCANBus

+ func (b *SocketCANBus) Name() string

+ func (b *SocketCANBus) ReadFrames(ctx context.Context, rxFrames chan<- RxFrame) error

+ func (b *SocketCANBus) WriteFrames(ctx context.Context, txFrames <-chan TxRequest) error

+ type TracingConfig struct

+ Enabled bool

+ Endpoint string

+ SampleRatio float64

+ ServiceName string

+ ServiceVersion string

type TxRequest

+ Bus string

type ValueStore

+ func (vs *ValueStore) RemoveSource(bus string, source uint8)

+ type VirtualDevice struct

+ type VirtualDeviceConfig struct

+ NAME uint64

+ ProductInfo VirtualProductInfo

+ type VirtualDeviceManager struct

+ func NewVirtualDeviceManager(txFunc func(TxRequest), registry *DeviceRegistry, logger *slog.Logger, ...) *VirtualDeviceManager

+ func (m *VirtualDeviceManager) Add(cfg VirtualDeviceConfig)

+ func (m *VirtualDeviceManager) ClaimedSource() (uint8, bool)

+ func (m *VirtualDeviceManager) Devices() []VirtualDeviceStatus

+ func (m *VirtualDeviceManager) HandleBusClaim(source uint8, name uint64) bool

+ func (m *VirtualDeviceManager) HandleISORequest(dst uint8, requestedPGN uint32, requesterSrc uint8)

+ func (m *VirtualDeviceManager) Heartbeat()

+ func (m *VirtualDeviceManager) ProductInfoPayload(source uint8) []byte

+ func (m *VirtualDeviceManager) Ready() bool

+ func (m *VirtualDeviceManager) StartAfterDiscovery(ctx context.Context, delay time.Duration)

+ type VirtualDeviceStatus struct

+ ModelID string

+ NAME string

+ Source uint8

+ State string

+ type VirtualProductInfo struct

+ ModelID string

+ ModelSerial string

+ ModelVersion string

+ ProductCode uint16

+ SoftwareVersion string

v0.3.1

Mar 10, 2026

Changes in this version

+ const DefaultLagCheckInterval

+ const DefaultMaxFrameRate

+ const DefaultMaxLiveLag

+ const DefaultMinLagReconnectInterval

+ const DefaultRateBurst

+ func HealthHandler(cfg HealthConfig) http.HandlerFunc

+ func MetricsHandler(broker *Broker, replStatus func() *ReplicationStatus) http.HandlerFunc

type Broker

+ func (b *Broker) LastFrameTime() time.Time

+ func (b *Broker) SendISORequest(dst uint8, pgn uint32) error

+ func (b *Broker) Stats() BrokerStats

+ type BrokerHealth struct

+ DeviceCount int

+ FramesTotal uint64

+ HeadSeq uint64

+ LastFrameTime time.Time

+ RingCapacity int

+ RingEntries uint64

+ Status string

+ type BrokerStats struct

+ ActiveConsumers int

+ ActiveSessions int

+ ActiveSubscribers int

+ DeviceCount int

+ FramesTotal uint64

+ HeadSeq uint64

+ LastFrameTime time.Time

+ RingCapacity int

+ RingEntries uint64

+ type BusSilenceMonitor struct

+ func NewBusSilenceMonitor(timeout time.Duration, broker *Broker, logger *slog.Logger) *BusSilenceMonitor

+ func (m *BusSilenceMonitor) IsSilent() bool

+ func (m *BusSilenceMonitor) Run(ctx context.Context)

+ type ByteMaskDiff struct

+ func (ByteMaskDiff) Apply(prev, diff []byte) []byte

+ func (ByteMaskDiff) Diff(prev, curr []byte) (bool, []byte)

+ type ChangeEvent struct

+ Data []byte

+ PGN uint32

+ Seq uint64

+ Source uint8

+ SubKey uint64

+ Timestamp time.Time

+ Type ChangeEventType

+ type ChangeEventType uint8

+ const Delta

+ const Idle

+ const Snapshot

+ func (t ChangeEventType) String() string

+ type ChangeReplayer struct

+ func NewChangeReplayer(methods map[uint32]DiffMethod, subKeys map[uint32]SubKeyFunc) *ChangeReplayer

+ func (r *ChangeReplayer) Apply(event ChangeEvent) ([]byte, error)

+ func (r *ChangeReplayer) State(source uint8, pgnID uint32, subKey uint64) []byte

+ type ChangeTracker struct

+ func NewChangeTracker(cfg ChangeTrackerConfig) *ChangeTracker

+ func (ct *ChangeTracker) Process(ts time.Time, source uint8, pgnID uint32, data []byte, seq uint64) *ChangeEvent

+ func (ct *ChangeTracker) Remove(source uint8, pgnID uint32, subKey uint64)

+ func (ct *ChangeTracker) Reset()

+ func (ct *ChangeTracker) Tick(now time.Time) []ChangeEvent

+ func (ct *ChangeTracker) TrackedPairs() int

+ type ChangeTrackerConfig struct

+ DefaultIdleTimeout time.Duration

+ DefaultMethod DiffMethod

+ IdleMultiplier int

+ IdleTimeouts map[uint32]time.Duration

+ Methods map[uint32]DiffMethod

+ SubKeys map[uint32]SubKeyFunc

+ type ComponentHealth struct

+ Message string

+ Status string

+ type DecodedDeviceValues struct

+ Manufacturer string

+ ModelID string

+ Name string

+ Source uint8

+ Values []DecodedPGNValue

+ type DecodedPGNValue struct

+ Description string

+ Fields any

+ PGN uint32

+ Seq uint64

+ Ts string

+ type DiffMethod interface

+ Apply func(prev, diff []byte) []byte

+ Diff func(prev, curr []byte) (significant bool, diff []byte)

type EventFilter

+ ExcludeNames []uint64

+ ExcludePGNs []uint32

+ type FieldTolerance struct

+ Field string

+ Tolerance float64

+ type FieldToleranceDiff struct

+ Decode func([]byte) (any, error)

+ Inner DiffMethod

+ PGN uint32

+ Tolerances []FieldTolerance

+ func (f *FieldToleranceDiff) Apply(prev, diff []byte) []byte

+ func (f *FieldToleranceDiff) Diff(prev, curr []byte) (bool, []byte)

+ type HealthConfig struct

+ Broker *Broker

+ BusSilenceThreshold time.Duration

+ ReplStatus func() *ReplicationStatus

+ type HealthStatus struct

+ Broker BrokerHealth

+ Components map[string]ComponentHealth

+ Replication *ReplicationHealth

+ Status string

type InstanceManager

+ func (im *InstanceManager) SetJournalRotation(duration time.Duration, size int64)

+ type PGNMatcher struct

+ Ranges [][2]uint32

+ Singles []uint32

+ func (m *PGNMatcher) Contains(pgn uint32) bool

type ReplicationClientConfig

+ LagCheckInterval int

+ MaxLiveLag uint64

+ MinLagReconnectInterval time.Duration

+ type ReplicationHealth struct

+ BackfillRemaining uint64

+ Connected bool

+ LastAck time.Time

+ LiveLag uint64

+ Status string

type ReplicationServer

+ MaxFrameRate float64

+ MaxLiveLag uint64

+ RateBurst int

+ type SendPolicy struct

+ Enabled bool

+ Rules []SendRule

+ type SendRule struct

+ Allow bool

+ Names []uint64

+ PGNs *PGNMatcher

+ func ParseSendRule(s string) (SendRule, error)

+ func ParseSendRules(rules []string) ([]SendRule, error)

+ func (r *SendRule) Matches(pgn uint32, dstNAME uint64, nameKnown bool) bool

+ type SubKeyFunc func(data []byte) uint64

type ValueStore

+ func (vs *ValueStore) DecodedSnapshot(devices *DeviceRegistry, filter *EventFilter) []DecodedDeviceValues

+ func (vs *ValueStore) DecodedSnapshotJSON(devices *DeviceRegistry, filter *EventFilter) json.RawMessage

v0.2.0

Mar 5, 2026

Changes in this version

+ var BuildCANID = canbus.BuildCANID

+ var ErrFallenBehind = errors.New("consumer fallen behind: data no longer available")

+ var ParseCANID = canbus.ParseCANID

+ func CANReader(ctx context.Context, iface string, rxFrames chan<- RxFrame, ...) error

+ func CANWriter(ctx context.Context, iface string, txFrames <-chan TxRequest, ...) error

+ func FragmentFastPacket(data []byte, seqCounter uint8) [][]byte

+ func IsFastPacket(pgn uint32) bool

+ func ParseISO8601Duration(s string) (time.Duration, error)

+ type ArchiveTrigger int

+ const ArchiveBeforeExpire

+ const ArchiveDisabled

+ const ArchiveOnRotate

+ func ParseArchiveTrigger(s string) (ArchiveTrigger, error)

+ func (t ArchiveTrigger) String() string

+ type BlockWriter struct

+ func NewBlockWriter(cfg BlockWriterConfig) (*BlockWriter, error)

+ func (w *BlockWriter) AppendBlock(baseSeq uint64, baseTimeUs int64, data []byte, compressed bool) error

+ func (w *BlockWriter) Close() error

+ type BlockWriterConfig struct

+ BlockSize int

+ Compression journal.CompressionType

+ Dir string

+ Logger *slog.Logger

+ OnRotate func(RotatedFile)

+ Prefix string

+ RotateDuration time.Duration

+ RotateSize int64

+ type Broker struct

+ func NewBroker(cfg BrokerConfig) *Broker

+ func (b *Broker) AckSession(id string, seq uint64) error

+ func (b *Broker) CloseRx()

+ func (b *Broker) CreateSession(id string, bufferTimeout time.Duration, filter *EventFilter) (*ClientSession, uint64)

+ func (b *Broker) CurrentSeq() uint64

+ func (b *Broker) Devices() *DeviceRegistry

+ func (b *Broker) Done() <-chan struct{}

+ func (b *Broker) GetSession(id string) *ClientSession

+ func (b *Broker) NewConsumer(cfg ConsumerConfig) *Consumer

+ func (b *Broker) Run()

+ func (b *Broker) RxFrames() chan<- RxFrame

+ func (b *Broker) SetJournal(ch chan<- RxFrame)

+ func (b *Broker) Subscribe(filter *EventFilter) (*subscriber, func())

+ func (b *Broker) TouchSession(id string)

+ func (b *Broker) TxFrames() <-chan TxRequest

+ func (b *Broker) Values() *ValueStore

+ type BrokerConfig struct

+ InitialHead uint64

+ JournalDir string

+ Logger *slog.Logger

+ MaxBufferDuration time.Duration

+ ReplicaMode bool

+ RingSize int

+ type CANHeader = canbus.CANHeader

+ type ClientSession struct

+ BufferTimeout time.Duration

+ Cursor uint64

+ Filter *EventFilter

+ ID string

+ LastActivity time.Time

+ type Consumer struct

+ func (c *Consumer) Close() error

+ func (c *Consumer) Cursor() uint64

+ func (c *Consumer) Next(ctx context.Context) (*Frame, error)

+ type ConsumerConfig struct

+ Cursor uint64

+ Filter *EventFilter

+ type Device struct

+ ByteCount uint64

+ DeviceClass uint8

+ DeviceFunction uint8

+ DeviceInstance uint8

+ FirstSeen time.Time

+ LastSeen time.Time

+ Manufacturer string

+ ManufacturerCode uint16

+ ModelID string

+ ModelSerial string

+ ModelVersion string

+ NAME uint64

+ NAMEHex string

+ PacketCount uint64

+ ProductCode uint16

+ SoftwareVersion string

+ Source uint8

+ UniqueNumber uint32

+ type DeviceRegistry struct

+ func NewDeviceRegistry() *DeviceRegistry

+ func (r *DeviceRegistry) Get(source uint8) *Device

+ func (r *DeviceRegistry) HandleAddressClaim(source uint8, data []byte) *Device

+ func (r *DeviceRegistry) HandleProductInfo(source uint8, data []byte) *Device

+ func (r *DeviceRegistry) RecordPacket(source uint8, ts time.Time, dataLen int) bool

+ func (r *DeviceRegistry) Snapshot() []Device

+ func (r *DeviceRegistry) SnapshotJSON() json.RawMessage

+ func (r *DeviceRegistry) SynthesizeFrames(ts time.Time) []RxFrame

+ type DeviceValues struct

+ Manufacturer string

+ ModelID string

+ Name string

+ Source uint8

+ Values []PGNValue

+ type EventFilter struct

+ Instances []uint8

+ Manufacturers []string

+ Names []uint64

+ PGNs []uint32

+ func ParseFilterParams(r *http.Request) (*EventFilter, error)

+ func (f *EventFilter) IsEmpty() bool

+ type EventLog struct

+ func NewEventLog() *EventLog

+ func (l *EventLog) Recent(n int) []ReplicationEvent

+ func (l *EventLog) Record(typ ReplicationEventType, detail map[string]any)

+ type FastPacketAssembler struct

+ func NewFastPacketAssembler(timeout time.Duration) *FastPacketAssembler

+ func (a *FastPacketAssembler) Process(pgn uint32, source uint8, data []byte, now time.Time) []byte

+ func (a *FastPacketAssembler) PurgeStale(now time.Time)

+ type Frame struct

+ Data []byte

+ Header CANHeader

+ Seq uint64

+ Timestamp time.Time

+ func (f *Frame) JSON() ([]byte, error)

+ type HoleTracker struct

+ func NewHoleTracker() *HoleTracker

+ func (h *HoleTracker) Add(start, end uint64)

+ func (h *HoleTracker) ContinuousThrough(cursor uint64) uint64

+ func (h *HoleTracker) Fill(start, end uint64) bool

+ func (h *HoleTracker) Holes() []SeqRange

+ func (h *HoleTracker) Len() int

+ func (h *HoleTracker) TotalMissing() uint64

+ type InstanceManager struct

+ func NewInstanceManager(dataDir string, logger *slog.Logger) (*InstanceManager, error)

+ func (im *InstanceManager) Get(id string) *InstanceState

+ func (im *InstanceManager) GetOrCreate(id string) *InstanceState

+ func (im *InstanceManager) List() []InstanceSummary

+ func (im *InstanceManager) SetInstancePaused(instanceID string, paused bool)

+ func (im *InstanceManager) SetOnRotate(fn func(instanceID string, rf RotatedFile))

+ func (im *InstanceManager) Shutdown()

+ type InstanceState struct

+ BoatHeadSeq uint64

+ BoatJournalBytes uint64

+ Connected bool

+ Cursor uint64

+ HoleTracker *HoleTracker

+ ID string

+ LastSeen time.Time

+ PersistedHoles []SeqRange

+ func (s *InstanceState) RecentEvents(n int) []ReplicationEvent

+ func (s *InstanceState) RecordEvent(typ ReplicationEventType, detail map[string]any)

+ func (s *InstanceState) Status() InstanceStatus

+ type InstanceStatus struct

+ BoatHeadSeq uint64

+ BoatJournalBytes uint64

+ Connected bool

+ Cursor uint64

+ Holes []SeqRange

+ ID string

+ LagSeqs uint64

+ LastSeen time.Time

+ type InstanceSummary struct

+ BoatHeadSeq uint64

+ Connected bool

+ Cursor uint64

+ Holes int

+ ID string

+ LagSeqs uint64

+ LastSeen time.Time

+ type JournalConfig struct

+ BlockSize int

+ Compression journal.CompressionType

+ Dir string

+ Logger *slog.Logger

+ OnRotate func(RotatedFile)

+ Prefix string

+ RotateCount int64

+ RotateDuration time.Duration

+ RotateSize int64

+ type JournalKeeper struct

+ func NewJournalKeeper(cfg KeeperConfig) *JournalKeeper

+ func (k *JournalKeeper) Run(ctx context.Context)

+ func (k *JournalKeeper) Send(rf RotatedFile)

+ func (k *JournalKeeper) SetOnPauseChange(fn func(dir KeeperDir, paused bool))

+ type JournalWriter struct

+ func NewJournalWriter(cfg JournalConfig, devices *DeviceRegistry, ch <-chan RxFrame) (*JournalWriter, error)

+ func (w *JournalWriter) Run(ctx context.Context) error

+ func (w *JournalWriter) SetPaused(p bool)

+ type KeeperConfig struct

+ ArchiveCommand string

+ ArchiveTrigger ArchiveTrigger

+ DirFunc func() []KeeperDir

+ Dirs []KeeperDir

+ Logger *slog.Logger

+ MaxAge time.Duration

+ MaxSize int64

+ MinKeep time.Duration

+ OnPauseChange func(dir KeeperDir, paused bool)

+ OverflowPolicy OverflowPolicy

+ SoftPct int

+ type KeeperDir struct

+ Dir string

+ InstanceID string

+ type OverflowPolicy int

+ const OverflowDeleteUnarchived

+ const OverflowPauseRecording

+ func ParseOverflowPolicy(s string) (OverflowPolicy, error)

+ func (p OverflowPolicy) String() string

+ type PGNValue struct

+ Data string

+ PGN uint32

+ Seq uint64

+ Ts string

+ type ReplicationClient struct

+ func NewReplicationClient(cfg ReplicationClientConfig, broker *Broker) *ReplicationClient

+ func (c *ReplicationClient) Run(ctx context.Context) error

+ func (c *ReplicationClient) Status() ReplicationStatus

+ type ReplicationClientConfig struct

+ CAFile string

+ CertFile string

+ InstanceID string

+ KeyFile string

+ Logger *slog.Logger

+ Target string

+ type ReplicationEvent struct

+ Detail map[string]any

+ Time time.Time

+ Type ReplicationEventType

+ type ReplicationEventType string

+ const EventBackfillStart

+ const EventBackfillStop

+ const EventBlockReceived

+ const EventCheckpoint

+ const EventLiveStart

+ const EventLiveStop

+ type ReplicationServer struct

+ func NewReplicationServer(im *InstanceManager, logger *slog.Logger) *ReplicationServer

+ func (s *ReplicationServer) Backfill(stream pb.Replication_BackfillServer) error

+ func (s *ReplicationServer) GetInstanceBroker(instanceID string) *Broker

+ func (s *ReplicationServer) GetInstanceState(instanceID string) *InstanceState

+ func (s *ReplicationServer) Handshake(ctx context.Context, req *pb.HandshakeRequest) (*pb.HandshakeResponse, error)

+ func (s *ReplicationServer) Live(stream pb.Replication_LiveServer) error

+ type ReplicationStatus struct

+ BackfillRemainingSeqs uint64

+ CloudCursor uint64

+ Connected bool

+ Holes []SeqRange

+ InstanceID string

+ LastAck time.Time

+ LiveLag uint64

+ LocalHeadSeq uint64

+ type RotatedFile struct

+ InstanceID string

+ Path string

+ type RxFrame struct

+ Data []byte

+ Header CANHeader

+ Seq uint64

+ Timestamp time.Time

+ type SeqRange struct

+ End uint64

+ Start uint64

+ type Server struct

+ func NewServer(broker *Broker, logger *slog.Logger) *Server

+ func (s *Server) HandleEphemeralSSE(w http.ResponseWriter, r *http.Request)

+ func (s *Server) HandleFunc(pattern string, handler func(http.ResponseWriter, *http.Request))

+ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request)

+ type SyncState struct

+ BoatHeadSeq uint64

+ BoatJournalBytes uint64

+ Cursor uint64

+ Holes []SeqRange

+ type TxRequest struct

+ Data []byte

+ Header CANHeader

+ type ValueStore struct

+ func NewValueStore() *ValueStore

+ func (vs *ValueStore) Record(source uint8, pgn uint32, ts time.Time, data []byte, seq uint64)

+ func (vs *ValueStore) Snapshot(devices *DeviceRegistry, filter *EventFilter) []DeviceValues

+ func (vs *ValueStore) SnapshotJSON(devices *DeviceRegistry, filter *EventFilter) json.RawMessage

v0.1.2

Mar 1, 2026

v0.1.1

Mar 1, 2026

v0.1.0

Mar 1, 2026

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL