model

package
v1.2.5 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Aug 8, 2026 License: Apache-2.0 Imports: 9 Imported by: 0

Documentation

Overview

Package model defines beacon's configuration entities: sources, sinks, and connectors. Validation here is structural; CEL filter compilation is checked by the filter package at apply time.

Index

Constants

View Source 
const (
	StreamFormatYDRaw     = "ydraw"     // Yacht Devices RAW ASCII line protocol
	StreamFormatActisense = "actisense" // Actisense binary stream protocol
)

Gateway stream formats (Source.Format, source types tcp/udp only).

View Source 
const (
	FileFormatNDJSON  = "ndjson"
	FileFormatCANDump = "candump"
)

File sink formats (Sink.Format, sink type file only).

View Source 
const (
	DefaultMaxFileBytes int64 = 100 << 20 // 100 MiB
	DefaultMaxFiles           = 5
)

Defaults applied by the file sink when MaxFileBytes/MaxFiles are left unset (0). MaxFiles counts the active file plus rotated backups, so the default keeps the active file and 4 rotated backups.

View Source 
const (
	DefaultHTTPPostBatchSize      = 100
	MaxHTTPPostBatchSize          = 1000
	DefaultHTTPPostRequestTimeout = Duration(10 * time.Second)
)

HTTP POST sink defaults and bounds. BatchSize is a maximum: a route sends a smaller batch immediately when fewer pending envelopes are available.

View Source 
const (
	DefaultPostgresTable        = "public.beacon_envelopes"
	DefaultPostgresBatchSize    = 100
	MaxPostgresBatchSize        = 1000
	DefaultPostgresWriteTimeout = Duration(10 * time.Second)
)

PostgreSQL sink defaults and bounds. PostgreSQL's bind-parameter limit is comfortably above MaxPostgresBatchSize times the sink's column count, while the cap keeps one confirmed write from becoming unreasonably large.

View Source 
const DefaultMaxMessages int64 = 10_000

DefaultMaxMessages is the retention cap used when a connector does not configure any buffer limit. Keep it exported so every configuration surface can show the same effective default instead of presenting an empty field.

Variables

View Source 
var ReservedPathPrefixes = []string{
	"/api", "/assets", "/cel-completions", "/config", "/connectors",
	"/dashboard", "/docs", "/frag", "/health", "/mcp", "/metrics",
	"/n2k", "/sinks", "/sources",
}

ReservedPathPrefixes cannot be used by HTTP sink paths.

Functions

func NormalizeMQTTBrokerURL 

func NormalizeMQTTBrokerURL(raw string) string

Types

type BridgeMode 

type BridgeMode string
const (
	BridgeSemantic    BridgeMode = "semantic"
	BridgeTransparent BridgeMode = "transparent"
	BridgeObserve     BridgeMode = "observe"
)

type BufferLimits 

type BufferLimits struct {
	MaxMessages int64    `json:"max_messages,omitempty"`
	MaxAge      Duration `json:"max_age,omitempty"`
	MaxBytes    int64    `json:"max_bytes,omitempty"`
}

func (BufferLimits) ApplyDefaults 

func (l BufferLimits) ApplyDefaults() BufferLimits

ApplyDefaults returns l with the spec default (max_messages=10000) applied when no limit at all is set.

type Config 

type Config struct {
	Sources    []Source    `json:"sources"`
	Sinks      []Sink      `json:"sinks"`
	Connectors []Connector `json:"connectors"`
}

func (*Config) Validate 

func (c *Config) Validate() error

Validate checks structural rules across the whole config: per-entity rules, ID uniqueness, reference integrity, and sink path collisions.

type Connector 

type Connector struct {
	ID                string       `json:"id"`
	Name              string       `json:"name"`
	SourceID          string       `json:"source_id"`
	SinkID            string       `json:"sink_id"`
	Filters           []string     `json:"filters,omitempty"`
	Buffer            BufferLimits `json:"buffer"`
	Enabled           bool         `json:"enabled"`
	Mode              BridgeMode   `json:"mode,omitempty"`
	ForwardManagement bool         `json:"forward_management,omitempty"`
}

func (Connector) EffectiveMode 

func (c Connector) EffectiveMode() BridgeMode

func (Connector) Validate 

func (c Connector) Validate() error

type Duration 

type Duration time.Duration

Duration marshals as a Go duration string ("90s", "24h").

func (Duration) MarshalJSON 

func (d Duration) MarshalJSON() ([]byte, error)

func (*Duration) UnmarshalJSON 

func (d *Duration) UnmarshalJSON(b []byte) error

type Sink 

type Sink struct {
	ID              string            `json:"id"`
	Name            string            `json:"name"`
	Type            SinkType          `json:"type"`
	Enabled         bool              `json:"enabled"`
	Interface       string            `json:"interface,omitempty"`         // socketcan
	Port            string            `json:"port,omitempty"`              // usbcan
	Path            string            `json:"path,omitempty"`              // http_sse / http_ws (served on data server)
	Address         string            `json:"address,omitempty"`           // tcp listen address; tcp_gateway: gateway host:port
	URL             string            `json:"url,omitempty"`               // mqtt broker, http_post endpoint, or postgres connection URL
	Topic           string            `json:"topic,omitempty"`             // mqtt
	Headers         map[string]string `json:"headers,omitempty"`           // http_post authentication and custom headers
	BatchSize       int               `json:"batch_size,omitempty"`        // http_post/postgres maximum envelopes per batch, 0 = default
	RequestTimeout  Duration          `json:"request_timeout,omitempty"`   // http_post request timeout, 0 = default
	Gzip            bool              `json:"gzip,omitempty"`              // http_post compress request bodies with gzip
	FilePath        string            `json:"file_path,omitempty"`         // file: absolute output path
	Format          string            `json:"format,omitempty"`            // file: "ndjson"/"candump"; tcp_gateway: "ydraw"/"actisense"
	MaxFileBytes    int64             `json:"max_file_bytes,omitempty"`    // file: rotate threshold, 0 = default
	MaxFiles        int               `json:"max_files,omitempty"`         // file: total files kept, 0 = default
	Table           string            `json:"table,omitempty"`             // postgres: schema-qualified destination table, blank = default
	AutoCreateTable bool              `json:"auto_create_table,omitempty"` // postgres: create/verify the destination schema at runtime
	TimescaleDB     bool              `json:"timescaledb,omitempty"`       // postgres: convert the table to a TimescaleDB hypertable
	WriteTimeout    Duration          `json:"write_timeout,omitempty"`     // postgres: schema/write timeout, 0 = default
}

func (Sink) EffectiveHTTPPostBatchSize added in v1.2.4 

func (s Sink) EffectiveHTTPPostBatchSize() int

func (Sink) EffectiveHTTPPostRequestTimeout added in v1.2.4 

func (s Sink) EffectiveHTTPPostRequestTimeout() time.Duration

func (Sink) EffectivePostgresBatchSize added in v1.2.5 

func (s Sink) EffectivePostgresBatchSize() int

func (Sink) EffectivePostgresTable added in v1.2.5 

func (s Sink) EffectivePostgresTable() string

func (Sink) EffectivePostgresWriteTimeout added in v1.2.5 

func (s Sink) EffectivePostgresWriteTimeout() time.Duration

func (Sink) Validate 

func (s Sink) Validate() error

type SinkType 

type SinkType string
const (
	SinkSocketCAN  SinkType = "socketcan"
	SinkUSBCAN     SinkType = "usbcan"
	SinkHTTPSSE    SinkType = "http_sse"
	SinkHTTPWS     SinkType = "http_ws"
	SinkHTTPPost   SinkType = "http_post" // POST confirmed JSON envelope batches to an HTTP(S) endpoint
	SinkTCP        SinkType = "tcp"       // TCP listener serving NDJSON to connecting clients
	SinkFile       SinkType = "file"
	SinkMQTT       SinkType = "mqtt"
	SinkPostgres   SinkType = "postgres"    // confirmed envelope batches persisted to PostgreSQL / TimescaleDB
	SinkTCPGateway SinkType = "tcp_gateway" // transmit onto an NMEA-2000 bus via a TCP gateway (YD / Actisense)
	SinkNull       SinkType = "null"        // accept and discard messages without external delivery
)

type Source 

type Source struct {
	ID        string            `json:"id"`
	Name      string            `json:"name"`
	Type      SourceType        `json:"type"`
	Enabled   bool              `json:"enabled"`
	Interface string            `json:"interface,omitempty"` // socketcan
	Port      string            `json:"port,omitempty"`      // usbcan (serial device path)
	URL       string            `json:"url,omitempty"`       // http_sse / http_ws / mqtt broker
	Topic     string            `json:"topic,omitempty"`     // mqtt
	Headers   map[string]string `json:"headers,omitempty"`   // http_sse / http_ws
	FilePath  string            `json:"file_path,omitempty"` // file: capture log to replay; gzip is transparent
	Address   string            `json:"address,omitempty"`   // tcp/udp: gateway host:port
	Format    string            `json:"format,omitempty"`    // tcp/udp: "ydraw" or "actisense"
}

func (Source) Validate 

func (s Source) Validate() error

type SourceType 

type SourceType string
const (
	SourceSocketCAN SourceType = "socketcan"
	SourceUSBCAN    SourceType = "usbcan"
	SourceHTTPSSE   SourceType = "http_sse"
	SourceHTTPWS    SourceType = "http_ws"
	SourceMQTT      SourceType = "mqtt"
	SourceFile      SourceType = "file" // replay an NMEA-2000 capture log (candump/canboat/YD/Actisense)
	SourceTCP       SourceType = "tcp"  // ingest from a TCP NMEA-2000 gateway (Yacht Devices / Actisense)
	SourceUDP       SourceType = "udp"  // ingest from a UDP NMEA-2000 gateway
)

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.