rtps

Documentation

Overview

Package rtps provides a pure-Go RTPS/UDP implementation of the dds interfaces. It requires no CGo and no installed system libraries; it speaks the OMG RTPS 2.3 wire protocol over UDP multicast/unicast so it can interoperate with any standards-compliant DDS implementation.

Create a participant with rtps.New. For reliable delivery, pass dds.ReliableQoS to NewPublisher/NewSubscriber. For payload-level security, pass rtps.WithSecurity to rtps.New:

p, err := rtps.New(dds.Domain(0), rtps.WithSecurity(myPlugin))

Package rtps provides a pure-Go RTPS/UDP implementation of the dds interfaces. It does not use CGo and runs on any platform that supports UDP sockets.

Two participants in the same process or on different hosts in the same LAN can exchange samples via standard RTPS wire protocol (OMG spec, version 2.3). Use the mock package for in-process testing; use this package when real network transport or interoperability with other DDS implementations is required.

Port Assignment

Ports follow the RTPS 2.3 §9.6.1 formula:

metaMulticast(domain) = 7400 + 250*domain
metaUnicast(domain,i) = 7400 + 250*domain + 10 + 2*i
dataUnicast(domain,i) = 7400 + 250*domain + 11 + 2*i

Discovery uses the multicast group 239.255.0.1.

Current limitations

• Remote subscription announcement (SEDP outbound reader) is not yet sent; remote writers will not push data to local readers across process boundaries unless both sides declare matching writers/readers for the same topic. Intra-process pub/sub works without SEDP matching.
• IPv6 transport is supported via WithIPv6 but has had limited interop testing.
• Large payloads are not fragmented (RTPS DATA_FRAG not implemented).

Index

• Constants
• Variables
• func New(domain dds.Domain, opts ...Option) (dds.Participant, error)
• func TopicMatches(pattern, topic string) bool
• type AckNack
• type DataFrag
• type DataSubmessage
• type EntityId
  • func (e EntityId) String() string
• type GUID
  • func (g GUID) String() string
• type Gap
• type GuidPrefix
  • func (p GuidPrefix) String() string
• type Header
• type Heartbeat
• type Locator
• type Option
  • func WithDeadlineCallback(fn func(topic string)) Option
  • func WithIPv6() Option
  • func WithNoMulticast() Option
  • func WithPeerLocators(addrs ...string) Option
  • func WithPersistentHistory(dir string) Option
  • func WithSecurity(plugin SecurityPlugin) Option
• type SecurityPlugin
• type SequenceNumber

Constants

const (
	EndpointSPDPAnnouncer    uint32 = 1 << 0
	EndpointSPDPDetector     uint32 = 1 << 1
	EndpointSEDPPubAnnouncer uint32 = 1 << 2
	EndpointSEDPPubDetector  uint32 = 1 << 3
	EndpointSEDPSubAnnouncer uint32 = 1 << 4
	EndpointSEDPSubDetector  uint32 = 1 << 5
)

Builtin endpoint availability bitmask (§8.5.4.3 / §9.6.2.2).

const (
	LocatorKindInvalid = -1
	LocatorKindUDPv4   = 1
	LocatorKindUDPv6   = 2
)

Variables

var (
	EntityIdParticipant        = EntityId{0x00, 0x00, 0x01, 0xC1}
	EntityIdSPDPWriter         = EntityId{0x00, 0x01, 0x00, 0xC2}
	EntityIdSPDPReader         = EntityId{0x00, 0x01, 0x00, 0xC7}
	EntityIdSEDPPubWriter      = EntityId{0x00, 0x00, 0x03, 0xC2}
	EntityIdSEDPPubReader      = EntityId{0x00, 0x00, 0x03, 0xC7}
	EntityIdSEDPSubWriter      = EntityId{0x00, 0x00, 0x04, 0xC2}
	EntityIdSEDPSubReader      = EntityId{0x00, 0x00, 0x04, 0xC7}
	EntityIdUnknown            = EntityId{0x00, 0x00, 0x00, 0x00}
	EntityIdBuiltinParticipant = EntityId{0x00, 0x00, 0x01, 0xC1}
)

Well-known entity IDs per RTPS 2.3 Table 9.1.

Functions

func New

func New(domain dds.Domain, opts ...Option) (dds.Participant, error)

New creates an RTPS participant joined to the given DDS domain. It binds UDP sockets, starts SPDP/SEDP, and returns a dds.Participant.

func TopicMatches

func TopicMatches(pattern, topic string) bool

TopicMatches reports whether pattern (which may contain MQTT-style + and # wildcards) matches the concrete topic name.

Rules:

• '+' matches exactly one topic level (no slashes).
• '#' at the end of a segment matches zero or more remaining levels.
• Literal segments must match exactly (case-sensitive).
• "foo/" and "foo" are distinct topics (two levels vs one level).

Types

type AckNack

type AckNack struct {
	ReaderEntityId EntityId
	WriterEntityId EntityId
	Base           SequenceNumber // first missing sequence number
	Bitmap         uint32         // bit N set → Base+N is missing
	Count          int32
}

AckNack holds the parsed fields of an ACKNACK submessage. We use a fixed 32-bit bitmap (NumBits always 32, one bitmap word).

type DataFrag

type DataFrag struct {
	WriterEntityId      EntityId
	ReaderEntityId      EntityId
	WriterSeqNum        SequenceNumber
	FragmentStartingNum uint32 // 1-based index of the first fragment in this submessage
	FragmentsInSubmsg   uint16 // number of fragments in this submessage
	FragmentSize        uint16 // size of each fragment in bytes (last may be smaller)
	DataSize            uint32 // total (unfragmented) data size in bytes
	Payload             []byte // raw bytes of the fragment(s)
}

DataFrag carries the fields of an RTPS DATA_FRAG submessage.

type DataSubmessage

type DataSubmessage struct {
	ReaderEntityId EntityId
	WriterEntityId EntityId
	SeqNum         SequenceNumber
	Payload        []byte // nil when Data flag not set
}

DataSubmessage holds the parsed fields of a DATA submessage.

type EntityId

type EntityId [4]byte

EntityId identifies a specific endpoint within a participant (§9.3.1).

func (EntityId) String

func (e EntityId) String() string

type GUID

type GUID struct {
	Prefix GuidPrefix
	Entity EntityId
}

GUID globally identifies a DDS entity: participant + endpoint.

func (GUID) String

func (g GUID) String() string

type Gap

type Gap struct {
	ReaderEntityId EntityId
	WriterEntityId EntityId
	GapStart       SequenceNumber // first irrelevant SN (inclusive)
	GapEnd         SequenceNumber // last irrelevant SN (inclusive)
}

Gap indicates a contiguous range of sequence numbers that are permanently unavailable from this writer (evicted from history). Receiving a GAP tells the reader to advance its expected-SN pointer past the covered range.

type GuidPrefix

type GuidPrefix [12]byte

GuidPrefix is the 12-byte participant identifier (RTPS 2.3 §9.3.1).

func (GuidPrefix) String

func (p GuidPrefix) String() string

type Header

type Header struct {
	ProtocolVersion [2]byte // {major, minor}
	VendorId        [2]byte
	GuidPrefix      GuidPrefix
}

Header is the fixed 20-byte RTPS message header (§9.4.1).

type Heartbeat

type Heartbeat struct {
	ReaderEntityId EntityId
	WriterEntityId EntityId
	FirstSN        SequenceNumber // lowest SN still in the writer's history
	LastSN         SequenceNumber // highest SN sent so far
	Count          int32          // monotonically increasing per writer
}

Heartbeat holds the parsed fields of a HEARTBEAT submessage.

type Locator

type Locator struct {
	Kind    int32
	Port    uint32
	Address [16]byte
}

Locator_t: 24-byte transport endpoint address (RTPS 2.3 §9.3.2).

type Option

type Option func(*participant)

Option configures a Participant at creation time.

func WithDeadlineCallback

func WithDeadlineCallback(fn func(topic string)) Option

WithDeadlineCallback sets a function that is called when a publisher has not written for longer than its QoS.Deadline period.

func WithIPv6

func WithIPv6() Option

WithIPv6 enables the IPv6 multicast transport. When set, the participant binds an additional pair of IPv6 UDP sockets and joins the RTPS IPv6 discovery group (FF03::1). IPv4 sockets are still created so the participant is reachable from both IPv4 and IPv6 peers.

func WithNoMulticast

func WithNoMulticast() Option

WithNoMulticast disables SPDP multicast discovery. The participant will not join or send to 239.255.0.1; use WithPeerLocators to supply peers manually.

func WithPeerLocators

func WithPeerLocators(addrs ...string) Option

WithPeerLocators adds static peer unicast addresses for unicast-only discovery. Each address must be a host:port string parseable by net.ResolveUDPAddr.

func WithPersistentHistory

func WithPersistentHistory(dir string) Option

WithPersistentHistory configures TransientLocal durability to be backed by files in dir. On participant start the last sample for each topic is loaded from disk; on every Write the new last sample is flushed to disk. The file name for topic T is <dir>/topic-<sanitised(T)>.bin. Each file holds a 4-byte length prefix followed by the raw sample bytes.

The option is a no-op when dir is "". Failures during load are silently ignored (missing files are normal on first run); failures during flush are also silently ignored so that a write to a read-only directory does not block the caller.

func WithSecurity

func WithSecurity(plugin SecurityPlugin) Option

WithSecurity returns an Option that applies plugin to every payload transmitted and received by this participant. All peers that communicate with this participant must use the same plugin and key material.

type SecurityPlugin

type SecurityPlugin interface {
	Seal(plaintext []byte) ([]byte, error)
	Open(ciphertext []byte) ([]byte, error)
}

SecurityPlugin is satisfied by any type that can seal (encrypt / sign) and open (decrypt / verify) a DDS payload byte slice. Built-in implementations live in the security sub-package; NullPlugin (pass-through) is also available there for development and testing.

type SequenceNumber

type SequenceNumber struct {
	High int32
	Low  uint32
}

SequenceNumber is the 8-byte writer sequence number (§9.3.2).