navlink

package module
v0.9.4 Latest Latest
Warning

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

 Go to latest Published: Sep 9, 2026 License: MIT Imports: 22 Imported by: 0

README

Go 语言 VDA5050 MQTT 接入 SDK：一个 Client 完成主题、强类型入站回调、强类型出站发布。

报文结构来自 vda5050-types-go，navlink 不另维护一份 schema。

安装

go get github.com/kalifun/navlink@v0.9.4

需要 Go 1.25+。

快速开始

client, err := navlink.New(navlink.Config{
    Broker:    "tcp://localhost:1883",
    ClientID:  "master-control",
    Interface: "uagv",
    Version:   "v2",
})
if err != nil {
    log.Fatal(err)
}

client.OnState(func(ctx context.Context, env navlink.Envelope, st *state.State) error {
    fmt.Println(env.AGV.SerialNumber, st.LastNodeId)
    return nil
})

if err := client.Start(ctx); err != nil {
    log.Fatal(err)
}
defer client.Stop(ctx)

可运行示例：

go run ./examples/subscribe-state
go run ./examples/platform-wiring          # 协议 L1 事件 + 应用自定义事件
go run ./examples/dispatch-egress-sketch   # ClassifyPublish 三态

发布

headerId / orderUpdateId / actionId 由调用方填写。库只补 version、timestamp 和 topic。

res, err := client.AGV(mfr, sn).PublishOrder(ctx, ord)
switch navlink.ClassifyPublish(err) {
case navlink.PublishOutcomeAccepted:
    // MQTT QoS 握手成功（broker 收下），不是「车已接受 order」
    _ = res.Topic
case navlink.PublishOutcomeNotStarted:
    // 协议 ID 可以再用
case navlink.PublishOutcomeUncertain:
    // 发送后超时 / 取消：不要复用这组 ID
}

发布前默认做轻量校验（headerId == 0、空 orderId 等），可用 Config.OutboundValidation 调整。

官方瞬时动作有 helper：CancelOrder、StartPause / StopPause、InitPosition、StateRequest、FactsheetRequest，以及无参的 StartCharging / StopCharging。厂商自定义 actionType 仍走 PublishInstantActions。

可选 Config.InboundPolicy（例如 NewHeaderSequencePolicy()）在 envelope 上标注 Accept|Stale|Duplicate，默认不丢包。

Config.IdentityMapper 把 (manufacturer, serial) → robotID 写到 Envelope.RobotID。厂商扩展字段走 Config.Extensions → Envelope.Meta，见 extend/README.md。

同一套 Client 可对接真实 MQTT，或 testkit 里的 FakeBroker。

入站回调

OnState / OnConnection / OnTopic 等跑在入站 worker 上（保序、不堵 Paho）。handler 与 SDK 内部都必须很快返回：禁止 HTTP、Subscribe / Track、长锁、同步 Publish 等 PUBACK。慢活与订阅变更放到平台自己的 goroutine / 队列。

MQTT 入站投递（语义 B）：按 (manufacturer, serial) 分片有序；connection 与自定义 OnTopic 各有独立队列，互不饿死车态。同车 connection 风暴在投递前 合并为最新态。OnInboundDrop(topic, reason)：InboundDropped 为真丢（visualization 满队列）；InboundBackpressured 为背压（将阻塞等待，未丢）。

FleetSession 是可选助手：默认 不会因 ONLINE 自动订 state。需要时显式 Client.Track，或 opt-in FleetOptions.AutoTrackFromConnection（Track 在独立 worker，不堵 typed 入站）。边界说明见 docs/INBOUND_BOUNDARY.md。

Envelope.ReceivedAt 是报文入队时刻（Paho 回调），DispatchedAt 是 worker 开始处理的时刻，QueueWait() 是队列等待。不要把 ReceivedAt 当成「MQTT 慢」。可选 Config.SlowInbound + OnSlowInbound 在队列等待或 handler 过长时告警。

范围

navlink 只做 协议执行，不做：

  • headerId / orderUpdateId / actionId 的分配与水位
  • Uncertain 之后的 fencing / 换号重发
  • 选车、路径规划、交通管制、何时充电等业务判定
  • 何时订哪辆车的 state（平台策略；可用显式 Track）
  • 默认 Redis / 跨进程 EventBus、多租户网关
  • 规定消费方如何分层
  • 领域事件——需要的话用 Emit 自己挂

VDA 收发走 Client / AGV；其它应用 MQTT 走 Client.Transport()。

更多见 CHANGELOG.md。

开发

本仓库用 direnv + Nix（.envrc → github:kalifun/devshells#go-1_25）：

direnv allow
make test
make check   # generr + fmt + vet + test

错误码由 glitch 从 errors/*.yaml 生成：

make generr

请勿手改 internal/gerrors/ 下的生成文件。

License

MIT

Documentation

Overview

Package navlink is a VDA5050 protocol access SDK for Go.

It provides MQTT connectivity, a single TopicResolver, typed inbound handlers (OnState / OnConnection / …), and typed outbound publishing via AGVHandle. Scheduling and domain orchestration stay outside this package.

Index

Constants

View Source 
const (
	EventStateReceived         = "vda.state.received"
	EventConnectionChanged     = "vda.connection.changed"
	EventVisualizationReceived = "vda.visualization.received"
	EventFactsheetReceived     = "vda.factsheet.received"
	EventDecodeFailed          = "vda.decode.failed"
)

L1 protocol event names (stable; do not rename casually).

View Source 
const (
	ReasonHeaderIDZero      = "headerId_zero"
	ReasonOrderIDEmpty      = "orderId_empty"
	ReasonOrderUpdateIDZero = "orderUpdateId_zero"
	ReasonActionIDEmpty     = "actionId_empty"
	ReasonIdentityMismatch  = "identity_mismatch"
)

Outbound validation reason keys (metadata "reason" on OutboundValidationFailed).

View Source 
const DefaultOrderQoS byte = 1

DefaultOrderQoS is used for order / instantActions (and non-viz subscribe) when Config.QoS is nil.

View Source 
const MetaInboundDisposition = "navlink.inboundDisposition"

MetaInboundDisposition is set on Envelope.Meta when an InboundPolicy is configured.

Variables

This section is empty.

Functions

func IsOutboundValidationFailed 

func IsOutboundValidationFailed(err error) bool

IsOutboundValidationFailed reports a rejected bad outbound packet (not a broker failure).

func IsPublishBrokerRejected 

func IsPublishBrokerRejected(err error) bool

IsPublishBrokerRejected reports broker/token rejection after the wait completed.

func IsPublishCanceled 

func IsPublishCanceled(err error) bool

IsPublishCanceled reports context cancellation (not a timeout).

func IsPublishNotStarted 

func IsPublishNotStarted(err error) bool

IsPublishNotStarted reports ClientNotStarted (or transport not running).

func IsPublishQoSRejected 

func IsPublishQoSRejected(err error) bool

IsPublishQoSRejected reports an unsupported QoS level.

func IsPublishTimeout 

func IsPublishTimeout(err error) bool

IsPublishTimeout reports a publish wait timeout or context deadline. context.Canceled is not a timeout; use IsPublishCanceled.

func IsPublishValidationFailed 

func IsPublishValidationFailed(err error) bool

IsPublishValidationFailed reports light outbound validation rejection (bad packet). Distinct from broker reject — safe to fix the packet; usually not a same-ID retry case.

func MarkPublishAttempted 

func MarkPublishAttempted(err error) error

MarkPublishAttempted records that MQTT Publish was already invoked. Timeout / cancel on the wrapped error classify as Uncertain.

func PublishAccepted 

func PublishAccepted(err error) bool

PublishAccepted reports whether the MQTT broker accepted the publish for the configured QoS (token wait completed without error). Equivalent to err == nil. This is not vehicle-side order acceptance — that still comes from inbound state. Platforms should call RecordSuccessfulPublish only when this is true.

func QoSOf 

func QoSOf(q byte) *byte

QoSOf returns a pointer for Config.QoS. Passing 0 means real MQTT QoS 0.

Types

type AGVHandle 

type AGVHandle struct {
	// contains filtered or unexported fields
}

AGVHandle publishes typed outbound messages to one AGV. Callers must assign headerId / orderUpdateId / actionId before publish.

func (*AGVHandle) CancelOrder 

func (a *AGVHandle) CancelOrder(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

CancelOrder publishes a standard cancelOrder instantAction. actionID and headerID must be supplied by the caller (orchestration layer).

func (*AGVHandle) FactsheetRequest 

func (a *AGVHandle) FactsheetRequest(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

FactsheetRequest publishes a standard factsheetRequest instantAction. This action is defined by VDA5050 2.1.0; the library does not gate it on Config.Version.

func (*AGVHandle) InitPosition 

func (a *AGVHandle) InitPosition(ctx context.Context, headerID uint32, actionID string, p InitPositionParams) (PublishResult, error)

InitPosition publishes a standard initPosition instantAction.

func (*AGVHandle) PublishInstantActions 

func (a *AGVHandle) PublishInstantActions(ctx context.Context, ia *instant_actions.InstantActions) (PublishResult, error)

PublishInstantActions publishes instantActions. Caller owns HeaderId and actionIds. See PublishOrder for success/failure semantics and PublishResult usage.

func (*AGVHandle) PublishOrder 

func (a *AGVHandle) PublishOrder(ctx context.Context, o *order.Order) (PublishResult, error)

PublishOrder publishes an order. Caller owns HeaderId and OrderUpdateId. On success, PublishResult describes the bytes/topic actually sent; on failure result may be zero or partially filled (topic/payload prepared before transport).

err == nil (PublishAccepted) means the MQTT QoS handshake succeeded (broker accepted the publish). It does not mean the vehicle accepted the order — that is still observed via inbound state. Platforms may RecordSuccessfulPublish only on accepted publish; navlink never records for them.

func (*AGVHandle) StartCharging 

func (a *AGVHandle) StartCharging(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

StartCharging publishes the official startCharging instantAction (no parameters). Vendor-specific charging parameters are not part of this helper; use PublishInstantActions.

func (*AGVHandle) StartPause 

func (a *AGVHandle) StartPause(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

StartPause publishes a standard startPause instantAction.

func (*AGVHandle) StateRequest 

func (a *AGVHandle) StateRequest(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

StateRequest publishes a standard stateRequest instantAction.

func (*AGVHandle) StopCharging 

func (a *AGVHandle) StopCharging(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

StopCharging publishes the official stopCharging instantAction (no parameters).

func (*AGVHandle) StopPause 

func (a *AGVHandle) StopPause(ctx context.Context, headerID uint32, actionID string) (PublishResult, error)

StopPause publishes a standard stopPause instantAction.

type Client 

type Client struct {
	// contains filtered or unexported fields
}

Client is the VDA5050 protocol access entrypoint.

func New 

func New(cfg Config) (*Client, error)

New validates config and constructs a Client (does not connect).

func (*Client) AGV 

func (c *Client) AGV(manufacturer, serial string) *AGVHandle

AGV returns a per-vehicle outbound handle.

func (*Client) Connected 

func (c *Client) Connected() bool

Connected reports whether the transport currently has an MQTT connection.

func (*Client) Emit 

func (c *Client) Emit(event string, payload any) error

Emit publishes a custom (or L1) event. Requires an attached EventBus.

func (*Client) EventBus 

func (c *Client) EventBus() EventBus

EventBus returns the attached bus, if any.

func (*Client) OnAGVOffline 

func (c *Client) OnAGVOffline(h func(Identity))

OnAGVOffline registers a FleetSession offline hook (no-op when Fleet is disabled).

func (*Client) OnAGVOnline 

func (c *Client) OnAGVOnline(h func(Identity))

OnAGVOnline registers a FleetSession online hook (no-op when Fleet is disabled).

func (*Client) OnConnection 

func (c *Client) OnConnection(h ConnectionHandler)

OnConnection registers a typed connection handler.

func (*Client) OnFactsheet 

func (c *Client) OnFactsheet(h FactsheetHandler)

OnFactsheet registers a typed factsheet handler.

func (*Client) OnHandlerError 

func (c *Client) OnHandlerError(h HandlerErrorHandler)

OnHandlerError registers a callback when a typed/raw inbound handler returns an error.

func (*Client) OnState 

func (c *Client) OnState(h StateHandler)

OnState registers a typed state handler (may be called before Start). The handler runs on the inbound worker and must return quickly; see StateHandler. With an EventBus attached, the handler is registered on EventStateReceived.

func (*Client) OnSubscriptionsRestored 

func (c *Client) OnSubscriptionsRestored(h func(error))

OnSubscriptionsRestored registers a callback after reconnect subscription restore.

func (*Client) OnTopic 

func (c *Client) OnTopic(filter string, h TopicHandler)

OnTopic registers a raw/escape-hatch topic filter handler.

func (*Client) OnTransportDown 

func (c *Client) OnTransportDown(h func(error))

OnTransportDown registers a callback for unexpected transport disconnects.

func (*Client) OnTransportUp 

func (c *Client) OnTransportUp(h func())

OnTransportUp registers a callback after a successful connect or restore.

func (*Client) OnVisualization 

func (c *Client) OnVisualization(h VisualizationHandler)

OnVisualization registers a typed visualization handler.

func (*Client) RestoreFleet 

func (c *Client) RestoreFleet(ctx context.Context) error

RestoreFleet re-subscribes fleet connection and tracked AGVs after reconnect.

func (*Client) Start 

func (c *Client) Start(ctx context.Context) error

Start connects the transport and establishes subscriptions for registered handlers.

func (*Client) Stop 

func (c *Client) Stop(ctx context.Context) error

Stop cancels subscriptions and disconnects the transport.

func (*Client) Subscribe 

func (c *Client) Subscribe(event string, h EventHandler) (Unsubscribe, error)

Subscribe registers a handler for an event name (L1 or platform custom). Subscribing to an L1 event also marks the corresponding MQTT channel as wanted.

func (*Client) Topics 

func (c *Client) Topics() topic.Resolver

Topics returns the TopicResolver bound to this client.

func (*Client) Track 

func (c *Client) Track(ctx context.Context, manufacturer, serial string) error

Track manually tracks an AGV (subscribes per-AGV channels). Requires Fleet. Call from a platform goroutine/queue — not from an On* inbound handler.

func (*Client) Transport 

func (c *Client) Transport() Transport

Transport returns the byte-level transport. Use this for non-VDA application MQTT; do not mix raw traffic into AGV helpers.

func (*Client) Untrack 

func (c *Client) Untrack(ctx context.Context, manufacturer, serial string) error

Untrack stops per-AGV subscriptions. Requires Fleet.

func (*Client) UseEventBus 

func (c *Client) UseEventBus(bus EventBus)

UseEventBus attaches a bus and migrates any already-registered On* handlers onto it. After this call, On* registers as bus subscribers; inbound messages Publish to the bus.

type Config 

type Config struct {
	// Broker is the MQTT broker URL, e.g. tcp://localhost:1883.
	// Required unless Transport is provided.
	Broker string

	// ClientID is the MQTT client id. Required unless Transport is provided.
	ClientID string

	Username string
	Password string

	// Interface is the VDA topic prefix (e.g. uagv / vda5050). Required.
	Interface string
	// Version is the VDA topic version segment (e.g. v2 / v2.0.0). Required.
	Version string

	// HeaderVersion is written into outbound ProtocolHeader.Version for Order and
	// InstantActions. Empty means Version is used (same Client policy for both).
	HeaderVersion string

	// Manufacturer / SerialNumber optionally pin subscriptions to one AGV.
	// Empty means fleet-level `+` wildcards for channels with handlers.
	Manufacturer string
	SerialNumber string

	// QoS is the MQTT QoS for publishes and subscriptions.
	// nil = library defaults (order/instantActions publish 1; visualization subscribe 0).
	// A pointer to 0 is a real QoS 0 (not "unset").
	QoS *byte

	KeepAlive      time.Duration
	ConnectTimeout time.Duration
	CleanSession   bool
	AutoReconnect  bool
	TLS            *tls.Config
	Will           *LastWill

	// InboundQueueSize is each inbound lane/shard queue length (default 256).
	InboundQueueSize int
	// OnInboundDrop is called when an inbound queue is full.
	// InboundDropped: message discarded (visualization).
	// InboundBackpressured: callback will block until space (message not lost).
	OnInboundDrop func(topic string, reason InboundDropReason)

	// SlowInbound, if > 0, invokes OnSlowInbound when queue wait or handler
	// runtime meets the threshold. Handlers still run synchronously on the
	// inbound worker; this is observability only.
	SlowInbound   time.Duration
	OnSlowInbound func(env Envelope, cause InboundSlowCause, d time.Duration)

	// RestoreSubscriptionsOnReconnect re-subscribes VDA topics after MQTT reconnect.
	// Default true. Applies when the transport implements ReconnectAware (built-in MQTT,
	// testkit FakeBroker). FleetSession uses Restore; other subscriptions are recreated.
	RestoreSubscriptionsOnReconnect *bool

	// StrictIdentity validates payload manufacturer/serial against the topic (default true).
	StrictIdentity *bool

	// Transport injects a custom transport (tests / shared connection).
	// When nil, an MQTT transport is created from Broker settings.
	Transport Transport

	// Fleet enables fleet tracking when non-nil.
	// Pass &DefaultFleetOptions() or a customized FleetOptions.
	// An all-zero FleetOptions is treated as DefaultFleetOptions().
	Fleet *FleetOptions

	// Extensions fills Envelope.Meta from vendor fields (optional).
	Extensions *extend.Registry

	// Bus is an optional EventBus (same as Client.UseEventBus).
	Bus EventBus

	// IdentityMapper optionally fills Envelope.RobotID.
	IdentityMapper IdentityMapper

	// OutboundValidation configures light pre-publish checks. Nil = enabled defaults
	// (reject headerId 0, empty orderId, orderUpdateId 0, empty actionId, identity mismatch).
	OutboundValidation *OutboundValidation

	// InboundPolicy optionally classifies inbound headerId (Accept/Stale/Duplicate).
	// Nil = accept-all. Classification is annotated on Envelope; messages are not dropped.
	InboundPolicy InboundPolicy

	// OnDecodeError is called when decode or identity checks fail.
	OnDecodeError DecodeErrorHandler
	// OnHandlerError is called when a typed/raw handler returns an error.
	OnHandlerError  HandlerErrorHandler
	OnTransportUp   func()
	OnTransportDown func(error)
	// OnSubscriptionsRestored is called after reconnect restore (nil err = success).
	OnSubscriptionsRestored func(error)
}

Config configures a navlink Client.

type ConnectionEvent 

type ConnectionEvent struct {
	Envelope   Envelope
	Connection *connection.Connection
}

ConnectionEvent is the payload for EventConnectionChanged.

type ConnectionHandler 

type ConnectionHandler func(ctx context.Context, env Envelope, msg *connection.Connection) error

ConnectionHandler handles a decoded connection message. Same rules as StateHandler. Do not Client.Track / Subscribe here; use a platform queue or explicit Track outside the inbound path.

type ConnectionLostAware 

type ConnectionLostAware interface {
	SetOnConnectionLost(fn func(error))
}

ConnectionLostAware is implemented by transports that can signal unexpected disconnects.

type ConnectionStatus 

type ConnectionStatus interface {
	Connected() bool
}

ConnectionStatus reports whether the transport is currently connected.

type DecodeErrorHandler 

type DecodeErrorHandler func(env Envelope, err error)

DecodeErrorHandler observes decode/identity failures without crashing the process.

type DecodeFailedEvent 

type DecodeFailedEvent struct {
	Envelope Envelope
	Err      error
}

DecodeFailedEvent is the payload for EventDecodeFailed.

type Envelope 

type Envelope struct {
	AGV     Identity
	Topic   string
	Channel topic.Channel
	Raw     []byte
	// ReceivedAt is when the MQTT callback enqueued the payload (UTC).
	// Transports without a queue (FakeBroker) set it to the same instant as DispatchedAt.
	ReceivedAt time.Time
	// DispatchedAt is when the inbound worker started handling the message (UTC).
	DispatchedAt time.Time
	Header       HeaderSummary
	Meta         Meta
	RobotID      string // filled when Config.IdentityMapper is set

	// InboundDisposition is set when Config.InboundPolicy is configured.
	// Empty means unclassified (default accept-all).
	InboundDisposition InboundDisposition
}

Envelope is the inbound message shell around a typed VDA5050 payload.

func (Envelope) QueueWait added in v0.9.1 

func (e Envelope) QueueWait() time.Duration

QueueWait is DispatchedAt − ReceivedAt: time spent in the inbound queue. Zero if either timestamp is unset, or if the result would be negative.

type EventBus 

type EventBus interface {
	Publish(ctx context.Context, event string, payload any) error
	Subscribe(event string, h EventHandler) (Unsubscribe, error)
}

EventBus is the optional protocol/custom event surface.

func NewMemoryEventBus 

func NewMemoryEventBus() EventBus

NewMemoryEventBus returns a process-local synchronous EventBus.

type EventHandler 

type EventHandler func(ctx context.Context, payload any) error

EventHandler handles a bus event payload.

type FactsheetEvent 

type FactsheetEvent struct {
	Envelope  Envelope
	Factsheet *factsheet.Factsheet
}

FactsheetEvent is the payload for EventFactsheetReceived.

type FactsheetHandler 

type FactsheetHandler func(ctx context.Context, env Envelope, msg *factsheet.Factsheet) error

FactsheetHandler handles a decoded factsheet message.

type FleetOptions 

type FleetOptions struct {
	// SubscribeState subscribes per-AGV state topics when tracked (default true).
	SubscribeState bool
	// SubscribeVisualization subscribes per-AGV visualization when tracked.
	SubscribeVisualization bool
	// AutoTrackFromConnection, when true, Track/Untrack from connection state on a
	// dedicated worker (not the inbound typed path). Default false: prefer explicit
	// Client.Track from the platform. See docs/INBOUND_BOUNDARY.md.
	AutoTrackFromConnection bool
}

FleetOptions configures per-AGV subscriptions when Config.Fleet is set.

func DefaultFleetOptions 

func DefaultFleetOptions() FleetOptions

DefaultFleetOptions returns the recommended fleet defaults.

type HandlerErrorHandler 

type HandlerErrorHandler func(env Envelope, err error)

HandlerErrorHandler observes inbound handler errors without crashing the process.

type HeaderSequencePolicy 

type HeaderSequencePolicy struct {
	// contains filtered or unexported fields
}

HeaderSequencePolicy tracks the last accepted headerId per key and classifies equal as Duplicate and lower as Stale. It does not drop messages.

func NewHeaderSequencePolicy 

func NewHeaderSequencePolicy() *HeaderSequencePolicy

NewHeaderSequencePolicy returns a policy that updates watermarks on Accept only.

func (*HeaderSequencePolicy) Classify 

func (p *HeaderSequencePolicy) Classify(agv Identity, channel topic.Channel, headerID uint32) InboundDisposition

Classify implements InboundPolicy.

type HeaderSummary 

type HeaderSummary struct {
	HeaderID     uint32
	Timestamp    string
	Version      string
	Manufacturer string
	SerialNumber string
}

HeaderSummary is a small decode of common VDA5050 header fields.

type Identity 

type Identity struct {
	Manufacturer string
	SerialNumber string
}

Identity is the protocol-level AGV identity (manufacturer + serial).

func (Identity) String 

func (id Identity) String() string

String returns manufacturer/serial for logging.

type IdentityMapper 

type IdentityMapper func(mfr, serial string) string

IdentityMapper maps protocol identity (mfr+sn) to a platform robot ID. Injected by the platform on Config; fills Envelope.RobotID on inbound paths. navlink never invents robot IDs and does not reverse-map for outbound — callers still publish via AGV(mfr, sn).

type InboundDisposition 

type InboundDisposition string

InboundDisposition is a light headerId classification for packet acceptance. navlink does not accept/reject business semantics — platforms decide whether to drop. 

const (
	InboundAccept    InboundDisposition = "accept"
	InboundStale     InboundDisposition = "stale"
	InboundDuplicate InboundDisposition = "duplicate"
)

type InboundDropReason added in v0.9.4 

type InboundDropReason int

InboundDropReason explains why OnInboundDrop fired. 

const (
	// InboundDropped means the message was discarded (visualization when the
	// inbound queue is full).
	InboundDropped InboundDropReason = iota
	// InboundBackpressured means the queue is full; the MQTT callback will
	// block until space is available (or the transport stops). The message is
	// not discarded.
	InboundBackpressured
)

func (InboundDropReason) String added in v0.9.4 

func (r InboundDropReason) String() string

type InboundPolicy 

type InboundPolicy interface {
	Classify(agv Identity, channel topic.Channel, headerID uint32) InboundDisposition
}

InboundPolicy classifies inbound messages by headerId for one (mfr, sn, channel). Default (nil Config.InboundPolicy) is accept-all — no classification.

type InboundSlowCause added in v0.9.1 

type InboundSlowCause int

InboundSlowCause says which inbound phase exceeded Config.SlowInbound. 

const (
	// InboundSlowQueue: time in the MQTT inbound queue before the worker ran.
	InboundSlowQueue InboundSlowCause = iota
	// InboundSlowHandler: time spent in decode + On* / OnTopic.
	InboundSlowHandler
)

func (InboundSlowCause) String added in v0.9.1 

func (c InboundSlowCause) String() string

type InitPositionParams 

type InitPositionParams struct {
	X          float64
	Y          float64
	Theta      float64
	MapID      string
	LastNodeID string
}

InitPositionParams is the VDA5050 initPosition parameter set (2.0 / 2.1). Keys on the wire are x, y, theta, mapId, lastNodeId — there is no lastNodeSequenceId.

type LastWill 

type LastWill struct {
	Topic   string
	Payload []byte
	QoS     byte
	Retain  bool
}

LastWill is an MQTT last-will message (optional).

type Meta 

type Meta map[string]any

Meta holds vendor extension fields produced by ExtensionRegistry (P1+).

type OutboundValidation 

type OutboundValidation struct {
	// Disabled turns off all outbound validation.
	Disabled bool
	// AllowZeroHeaderID permits headerId == 0 (default false).
	AllowZeroHeaderID bool
	// SkipIdentityCheck skips manufacturer/serial vs AGVHandle checks (default false).
	SkipIdentityCheck bool
}

OutboundValidation configures light pre-publish checks (no ID allocation). Nil Config.OutboundValidation means checks are enabled with defaults.

type PublishOptions 

type PublishOptions struct {
	QoS    byte
	Retain bool
}

PublishOptions controls MQTT publish behaviour.

type PublishOutcome 

type PublishOutcome int

PublishOutcome is the execution-layer three-way result of a publish. It answers whether the caller-supplied protocol IDs may be reused. 

const (
	// PublishOutcomeAccepted: broker accepted the publish (QoS handshake).
	// IDs are consumed; platforms may RecordSuccessfulPublish.
	PublishOutcomeAccepted PublishOutcome = iota
	// PublishOutcomeNotStarted: the publish never became a deliverable MQTT packet.
	// IDs may be returned to the pool and reused.
	PublishOutcomeNotStarted
	// PublishOutcomeUncertain: the packet may already be in flight (typical: PUBACK timeout).
	// IDs must not be reused. Fencing / recovery is orchestration.
	PublishOutcomeUncertain
)

func ClassifyPublish 

func ClassifyPublish(err error) PublishOutcome

ClassifyPublish maps a publish error to Accepted / NotStarted / Uncertain. Prefer this on the orchestration path; IsPublish* predicates remain for logs.

func (PublishOutcome) String 

func (o PublishOutcome) String() string

type PublishResult 

type PublishResult struct {
	Topic         string
	Channel       topic.Channel
	Manufacturer  string
	SerialNumber  string
	QoS           byte
	Payload       []byte
	HeaderID      uint32
	OrderID       string
	OrderUpdateID uint32
	ActionIDs     []string
}

PublishResult is a summary of what was actually handed to Transport.Publish. It is for reconciliation / logging only: navlink does not allocate IDs and does not RecordSuccessfulPublish — the orchestration layer decides that from err.

type RawHandler 

type RawHandler func(ctx context.Context, topic string, payload []byte) error

RawHandler handles a transport-level message (topic + payload).

type ReconnectAware 

type ReconnectAware interface {
	SetOnReconnect(fn func())
}

ReconnectAware is implemented by transports that can signal reconnects. The handler is invoked after a successful reconnect, not on the initial connect.

type StateEvent 

type StateEvent struct {
	Envelope Envelope
	State    *state.State
}

StateEvent is the payload for EventStateReceived.

type StateHandler 

type StateHandler func(ctx context.Context, env Envelope, msg *state.State) error

StateHandler handles a decoded state message. It runs on the inbound worker and must return quickly: no HTTP, no Subscribe, no long locks, no waiting for MQTT Publish. Do slow work in another goroutine.

type TopicHandler 

type TopicHandler func(ctx context.Context, env Envelope) error

TopicHandler is the escape hatch for non-typed topic filters. Same threading rules as StateHandler. On the built-in MQTT transport, custom topics share an isolated lane (not the per-AGV state shards).

type Transport 

type Transport interface {
	Start(ctx context.Context) error
	Stop(ctx context.Context) error
	Publish(ctx context.Context, topic string, payload []byte, opts PublishOptions) error
	Subscribe(ctx context.Context, filter string, handler RawHandler) (Unsubscribe, error)
}

Transport is the byte-level pub/sub boundary (MQTT, memory, etc.). VDA typed APIs live on Client; raw MQTT must not be mixed into AGV helpers.

type Unsubscribe 

type Unsubscribe func(ctx context.Context) error

Unsubscribe cancels a subscription.

type VisualizationEvent 

type VisualizationEvent struct {
	Envelope      Envelope
	Visualization *visualization.Visualization
}

VisualizationEvent is the payload for EventVisualizationReceived.

type VisualizationHandler 

type VisualizationHandler func(ctx context.Context, env Envelope, msg *visualization.Visualization) error

VisualizationHandler handles a decoded visualization message.

Source Files

View all Source files

Directories

Path Synopsis
examples
dispatch-egress-sketch command
Dispatch egress sketch: GetNext → fill → Publish → Record / return / fence from ClassifyPublish.
 Dispatch egress sketch: GetNext → fill → Publish → Record / return / fence from ClassifyPublish.
platform-wiring command
subscribe-state command
internal
bus
gerrors
Code generated by glitch.
 Code generated by glitch.
mqtt
outbound
session

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.