iotdataplane

package
v1.7.0 Latest Latest
Warning

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

 Go to latest Published: Oct 6, 2026 License: MIT Imports: 27 Imported by: 0

README

IoT Data Plane

Parity grade: A · SDK aws-sdk-go-v2/service/iotdataplane@v1.35.4 · last audited 2026-09-18 (2dfc55a39)

Coverage

Metric Value
PARITY entries audited 11 (9 ok, 2 partial)
Feature families 1 (1 ok)
Known gaps 3
Deferred items 1
Resource leaks clean
Known gaps
  • UnsupportedDocumentEncodingException (HTTP 415, modeled for the three shadow ops) is never returned: six AWS sources (botocore model, SDK errors.go, IoT API reference, shadow REST/protocol/troubleshooting guides) give no trigger condition, so any validation would be a guess.
  • GetConnection omits disconnectReason/disconnectedSince/thingName/vpcEndpointId, and every live-session field for admin-registered clients with no broker session: no disconnect history, principal or VPC-endpoint modeling exists.
  • SendDirectMessage.timeout (and the HTTP 504 on a missing PUBACK) is read nowhere: the MQTTPublisher boundary has no ack-wait. Publish with no broker wired drops the message after a warning (intentional degradation).
Deferred
  • Chaos fault-injection paths (ChaosServiceName/ChaosOperations) -- not part of AWS wire surface, no parity concern.

More

Documentation

Overview

Package iotdataplane provides the IoT Data Plane HTTP API for publishing messages directly to MQTT topics.

Index

Constants

This section is empty.

Variables

View Source 
var ErrConnectionExists = errors.New("connection already exists")

ErrConnectionExists is returned when trying to register a clientID that is already connected.

View Source 
var ErrConnectionNotFound = errors.New("connection not found")

ErrConnectionNotFound is returned when DeleteConnection targets a clientID that is not currently tracked as connected. Wire error code "ResourceNotFoundException" (real AWS iotdataplane exception; confirmed modeled for DeleteConnection via aws-sdk-go-v2/service/iotdataplane/deserializers.go's awsRestjson1_deserializeOpErrorDeleteConnection case list).

View Source 
var ErrNilAppContext = errors.New("AppContext is required")

ErrNilAppContext is returned when Provider.Init is called with a nil AppContext.

View Source 
var ErrNoBroker = errors.New("no mqtt broker configured")

ErrNoBroker is returned when no MQTT broker has been wired.

View Source 
var ErrNoSnapshot = errors.New("backend does not support restore")

ErrNoSnapshot is returned when a backend does not support snapshot/restore.

View Source 
var ErrRequestTooLarge = errors.New("RequestEntityTooLargeException")

ErrRequestTooLarge is returned when a shadow document or direct-message payload exceeds its maximum allowed size. Wire error code "RequestEntityTooLargeException" (real AWS iotdataplane exception, modeled for UpdateThingShadow and SendDirectMessage; NOT modeled for Publish).

View Source 
var ErrRetainedMessageNotFound = errors.New("retained message not found")

ErrRetainedMessageNotFound is returned when no retained message exists for a topic.

View Source 
var ErrShadowNotFound = errors.New("shadow not found")

ErrShadowNotFound is returned when a thing shadow is not found.

View Source 
var ErrValidation = errors.New("InvalidRequestException")

ErrValidation is returned for invalid input parameters.

View Source 
var ErrVersionConflict = errors.New("ConflictException")

ErrVersionConflict is returned when a shadow update specifies a version that does not match the current shadow version (optimistic locking violation). The wire error code is "ConflictException" (real AWS iotdataplane exception name; see aws-sdk-go-v2/service/iotdataplane/types.ConflictException) -- there is no "VersionConflictException" in the real API.

Functions

This section is empty.

Types

type Connection 

type Connection struct {
	// Session is the broker's live session for the client, nil when it has none.
	Session     *SessionInfo `json:"-"`
	ConnectedAt time.Time    `json:"connectedAt"`
	ClientID    string       `json:"clientId"`
	SourceIP    string       `json:"sourceIp,omitempty"`
}

Connection represents a registered MQTT client connection.

type Handler 

type Handler struct {
	Backend StorageBackend
	// contains filtered or unexported fields
}

Handler is the Echo HTTP handler for IoT Data Plane operations.

func NewHandler 

func NewHandler(backend StorageBackend) *Handler

NewHandler creates a new IoT Data Plane Handler.

func (*Handler) BackendFor 

func (h *Handler) BackendFor(region string) StorageBackend

BackendFor returns the backend serving region: the home backend, or the sibling built on first use.

func (*Handler) ChaosOperations 

func (h *Handler) ChaosOperations() []string

ChaosOperations returns all operations that can be fault-injected.

func (*Handler) ChaosRegions 

func (h *Handler) ChaosRegions() []string

ChaosRegions returns all regions this IoT Data Plane instance handles.

func (*Handler) ChaosServiceName 

func (h *Handler) ChaosServiceName() string

ChaosServiceName returns the lowercase AWS service name for fault rule matching.

func (*Handler) EnableRegions 

func (h *Handler) EnableRegions(home string)

EnableRegions makes h, the home region's handler, serve every other region through per-region siblings.

func (*Handler) ExtractOperation 

func (h *Handler) ExtractOperation(c *echo.Context) string

ExtractOperation returns the operation name.

func (*Handler) ExtractResource 

func (h *Handler) ExtractResource(c *echo.Context) string

ExtractResource extracts the topic or thing name from the URL path.

func (*Handler) GetSupportedOperations 

func (h *Handler) GetSupportedOperations() []string

GetSupportedOperations returns the list of supported operations.

func (*Handler) Handler 

func (h *Handler) Handler() echo.HandlerFunc

Handler returns the Echo handler function for IoT Data Plane operations.

func (*Handler) MatchPriority 

func (h *Handler) MatchPriority() int

MatchPriority returns the routing priority for the IoT Data Plane handler.

func (*Handler) Name 

func (h *Handler) Name() string

Name returns the service name.

func (*Handler) RegionBackends 

func (h *Handler) RegionBackends() []StorageBackend

RegionBackends returns the home backend followed by every sibling built so far.

func (*Handler) RegionHandler 

func (h *Handler) RegionHandler(region string) *Handler

RegionHandler returns the handler serving region: h itself for the home region, else its sibling.

func (*Handler) Reset 

func (h *Handler) Reset()

Reset clears all handler state by delegating to the backend.

func (*Handler) Restore 

func (h *Handler) Restore(ctx context.Context, data []byte) error

Restore implements persistence.Persistable.

func (*Handler) RouteMatcher 

func (h *Handler) RouteMatcher() service.Matcher

RouteMatcher returns a function matching IoT Data Plane requests.

func (*Handler) Snapshot 

func (h *Handler) Snapshot(ctx context.Context) []byte

Snapshot implements persistence.Persistable; other regions ride in an additive "regions" key.

type InMemoryBackend 

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

InMemoryBackend implements the IoT Data Plane backend.

shadows and connections are "dirty" store.Table-backed collections (shadowEntry / connectionEntry carry no exported fields, so they can't round-trip a direct JSON marshal) and are NOT registered on registry -- persistence.go drives them through an ephemeral DTO registry instead. retainedMessages is a "clean" table (RetainedMessage already carries its own identity as a real field, Topic) and IS registered on registry, so persistence.go drives it through registry.SnapshotAll()/RestoreAll() directly. See store_setup.go for the full registration.

func NewInMemoryBackend 

func NewInMemoryBackend() *InMemoryBackend

NewInMemoryBackend creates a new InMemoryBackend.

func (*InMemoryBackend) AddConnectionInternal 

func (b *InMemoryBackend) AddConnectionInternal(clientID string)

AddConnectionInternal seeds a connected client ID for testing purposes.

func (*InMemoryBackend) AddShadowInternal 

func (b *InMemoryBackend) AddShadowInternal(thingName, shadowName string, document []byte)

AddShadowInternal seeds a shadow entry for testing purposes. The document is parsed to extract desired/reported state; if parsing fails or no state key is present, the whole document is treated as the desired state.

func (*InMemoryBackend) DeleteConnection 

func (b *InMemoryBackend) DeleteConnection(clientID string) error

DeleteConnection disconnects a tracked MQTT client connection. Returns ErrConnectionNotFound if clientID has no tracked connection (real AWS models ResourceNotFoundException for this op -- see ErrConnectionNotFound). ClientIDs beginning with '$' are rejected per AWS rules.

func (*InMemoryBackend) DeleteConnectionWithOptions 

func (b *InMemoryBackend) DeleteConnectionWithOptions(clientID string, cleanSession, preventWill bool) error

DeleteConnectionWithOptions is DeleteConnection plus the real cleanSession and preventWillMessage query flags, applied to the broker's live session.

func (*InMemoryBackend) DeleteThingShadow 

func (b *InMemoryBackend) DeleteThingShadow(thingName, shadowName string) ([]byte, error)

DeleteThingShadow removes the document for the named shadow of a thing. Per AWS docs, the response is an "empty response state document" -- only version and timestamp, no state/metadata/clientToken (see device-shadow-document.html #device-shadow-example-response-json).

The shadow row is tombstoned (kept with state cleared) rather than physically removed: AWS explicitly documents that deleting a shadow does not reset its version number, so a later UpdateThingShadow that recreates this shadow must continue incrementing from the pre-delete version.

func (*InMemoryBackend) GetConnection added in v1.2.0 

func (b *InMemoryBackend) GetConnection(clientID string) (*Connection, error)

GetConnection returns connection information for clientID. Returns ErrConnectionNotFound (real AWS models ResourceNotFoundException for GetConnection; confirmed via deserializers.go's awsRestjson1_deserializeOpErrorGetConnection case list) when clientID has no tracked connection.

gopherstack's only concept of "connected" is the connections table populated by the gopherstack-only RegisterConnection admin extension (see admin-only-extensions family in PARITY.md) -- there is no real MQTT broker session tracking backing this. Fields the real GetConnectionOutput carries but this backend has no genuine data for (cleanSession, disconnectReason, disconnectedSince, keepAliveDuration, sessionExpiry, sourcePort, targetIp, targetPort, thingName, vpcEndpointId) are deliberately left unset on the returned Connection/omitted from the JSON response rather than fabricated -- see handleGetConnection.

func (*InMemoryBackend) GetRetainedMessage 

func (b *InMemoryBackend) GetRetainedMessage(topic string) (*RetainedMessage, error)

GetRetainedMessage returns the retained message stored for the given topic. ErrRetainedMessageNotFound is returned when no retained message exists for the topic.

func (*InMemoryBackend) GetThingShadow 

func (b *InMemoryBackend) GetThingShadow(thingName, shadowName string) ([]byte, error)

GetThingShadow returns the shadow document for the named shadow of a thing.

func (*InMemoryBackend) ListConnections 

func (b *InMemoryBackend) ListConnections() []*Connection

ListConnections returns all registered connections sorted by ConnectedAt ascending.

func (*InMemoryBackend) ListNamedShadowsForThing 

func (b *InMemoryBackend) ListNamedShadowsForThing(thingName string) ([]string, error)

ListNamedShadowsForThing returns the sorted list of named shadow names for the given thing. The classic (unnamed) shadow is excluded from this list.

func (*InMemoryBackend) ListRetainedMessages 

func (b *InMemoryBackend) ListRetainedMessages() ([]*RetainedMessage, error)

ListRetainedMessages returns summaries of all retained messages, sorted by topic.

func (*InMemoryBackend) ListSubscriptions added in v1.2.0 

func (b *InMemoryBackend) ListSubscriptions(clientID string) ([]SubscriptionSummary, error)

ListSubscriptions validates that clientID is a tracked connection, mirroring GetConnection/DeleteConnection's not-found semantics (real AWS models ResourceNotFoundException for ListSubscriptions too), then reports the client's live MQTT subscriptions read straight off the mochi-mqtt broker's per-client session state (MQTTPublisher.ClientSubscriptions, implemented in services/iot/broker.go off cl.State.Subscriptions). gopherstack's own connections table (populated only via the admin-only RegisterConnection extension) is a distinct, weaker notion of "connected" than a real broker session -- a tracked clientID whose broker session the broker doesn't currently know about (never actually connected over MQTT, or no broker wired at all) genuinely has no subscriptions to report, so an empty list is returned honestly rather than fabricated.

func (*InMemoryBackend) ListThingsWithShadows 

func (b *InMemoryBackend) ListThingsWithShadows() []string

ListThingsWithShadows returns the sorted list of thing names that have at least one shadow.

func (*InMemoryBackend) Publish 

func (b *InMemoryBackend) Publish(
	topic string,
	payload []byte,
	qos int32,
	retain bool,
	props MQTT5Properties,
) error

Publish delivers a message to the given MQTT topic, along with props (the optional MQTT5 packet properties PublishInput accepts) so live MQTT5 subscribers receive them as real packet properties. If no broker is configured the call returns ErrNoBroker. The retain flag is forwarded to the broker so live subscribers receive RETAIN=1 and the broker maintains retention canonically.

func (*InMemoryBackend) RegisterConnection 

func (b *InMemoryBackend) RegisterConnection(clientID, sourceIP string) error

RegisterConnection adds a client connection to the backend. Returns ErrConnectionExists if the clientID is already registered. ClientIDs beginning with '$' are rejected per AWS rules.

func (*InMemoryBackend) Reset 

func (b *InMemoryBackend) Reset()

Reset clears all backend state, including shadows, connections, and retained messages.

func (*InMemoryBackend) Restore 

func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error

Restore deserialises backend state from a JSON snapshot.

func (*InMemoryBackend) SendDirectMessage added in v1.2.0 

func (b *InMemoryBackend) SendDirectMessage(
	clientID, topic string,
	payload []byte,
	qos int32,
	props MQTT5Properties,
) error

SendDirectMessage delivers payload to topic, after confirming clientID is a tracked connection (ErrConnectionNotFound otherwise, matching GetConnection/ListSubscriptions/DeleteConnection).

Real AWS SendDirectMessage delivers to the target client specifically -- "the receiving client does not need to subscribe to the topic". When the mochi-mqtt broker has a live session for clientID, this now delivers via MQTTPublisher.SendToClient (services/iot/broker.go, cl.WritePacket), which really does write straight to that one client's connection regardless of its subscriptions -- true per-client delivery, matching AWS. gopherstack's own connections table (populated only via the admin-only RegisterConnection extension) is a distinct, weaker notion of "connected" than a real broker session, though: a clientID tracked there but with no live broker session (e.g. registered but never actually connected over MQTT) falls back to broadcasting on topic via Publish, the same honest best-effort approximation used before this method could address individual clients -- any live subscriber of topic really does observe the message. This fallback is a documented, deliberate divergence from real per-client delivery -- see PARITY.md gaps.

func (*InMemoryBackend) SetBroker 

func (b *InMemoryBackend) SetBroker(broker MQTTPublisher)

SetBroker wires the MQTT broker for publishing (called during CLI startup).

func (*InMemoryBackend) Snapshot 

func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte

Snapshot serialises backend state to JSON.

func (*InMemoryBackend) StoreRetainedMessage 

func (b *InMemoryBackend) StoreRetainedMessage(topic string, payload []byte, qos int32, userProperties []byte) error

StoreRetainedMessage saves a retained MQTT message for the given topic. Calling this with an empty payload removes the retained message for that topic (per AWS docs: "Publishing an empty (null) payload with retain = true deletes the retained message identified by topic"). userProperties is the raw (already base64-decoded) MQTT5 user properties blob from the Publish call, or nil if none were supplied. When the cap is reached, the oldest entry (by LastModifiedTime) is evicted to make room, matching AWS LRU behaviour and preventing silent publish failures.

func (*InMemoryBackend) UpdateThingShadow 

func (b *InMemoryBackend) UpdateThingShadow(thingName, shadowName string, document []byte) ([]byte, error)

UpdateThingShadow merges the desired/reported state from document into the stored shadow. AWS merge semantics: null values on individual keys delete them; a null section wipes the entire section; missing sections are left unchanged. The state key is required. The version is incremented on every successful update. Returns the updated shadow response including delta, metadata, and echoed clientToken.

type MQTT5Properties added in v1.3.1 

type MQTT5Properties struct {
	ContentType            string
	ResponseTopic          string
	PayloadFormatIndicator string
	CorrelationData        []byte
	UserProperties         []MQTT5UserProperty
	MessageExpiry          int64
}

MQTT5Properties carries the optional MQTT5 packet properties that AWS's Publish and SendDirectMessage operations accept (contentType, correlationData, responseTopic, payloadFormatIndicator, userProperties -- plus messageExpiry, Publish-only) so an MQTTPublisher implementation can forward them to live subscribers as real MQTT5 packet properties. A zero value means the corresponding request field was not supplied.

type MQTT5UserProperty added in v1.3.1 

type MQTT5UserProperty struct {
	Key   string
	Value string
}

MQTT5UserProperty is one key/value pair decoded from PublishInput's or SendDirectMessageInput's userProperties JSON array, where each array element is a single-key JSON object -- e.g. [{"deviceName": "alpha"}] (see PublishInput.UserProperties doc comment, aws-sdk-go-v2/service/iotdataplane@v1.35.4/api_op_Publish.go).

type MQTTPublisher 

type MQTTPublisher interface {
	// Publish delivers a message to an MQTT topic; any connected subscriber
	// of that topic observes it.
	Publish(topic string, payload []byte, retain bool, qos byte) error

	// PublishWithProperties behaves like Publish but also attaches props as
	// real MQTT5 packet properties. Only a subscriber connected with MQTT
	// protocol version 5 observes them -- MQTT 3.1.1 has no wire
	// representation for packet properties, which AWS documents explicitly
	// for userProperties ("For MQTT 3.1.1 clients, user properties are
	// silently dropped", SendDirectMessageInput.UserProperties doc) and
	// which applies to every MQTT5 property by protocol design.
	PublishWithProperties(
		topic string,
		payload []byte,
		retain bool,
		qos byte,
		props MQTT5Properties,
	) error

	// ClientSubscriptions returns the topic-filter -> QoS map of every
	// subscription clientID currently holds on the broker, and whether the
	// broker currently has a live client connected with that ID. A connected
	// client with zero subscriptions returns a non-nil empty map and true; a
	// client the broker doesn't currently know about -- never connected, or
	// (in gopherstack) only ever registered through iotdataplane's
	// admin-only RegisterConnection extension without a real broker session
	// -- returns (nil, false).
	ClientSubscriptions(clientID string) (subs map[string]byte, connected bool)

	// SendToClient delivers payload on topic directly to clientID's live
	// broker connection, bypassing topic subscription matching entirely --
	// mirroring real AWS SendDirectMessage's documented "the receiving
	// client does not need to subscribe to the topic" semantics. ok is false
	// when the broker has no live client with that ID: nothing was sent and
	// err is nil in that case, so callers can distinguish "no live
	// per-client route" from a genuine delivery failure.
	SendToClient(clientID, topic string, payload []byte, qos byte) (ok bool, err error)

	// ClientSession reports the live (not closed) session for clientID, or
	// ok=false when the broker has none.
	ClientSession(clientID string) (info SessionInfo, ok bool)

	// DisconnectClient closes clientID's live connection, optionally dropping session state and
	// its Last Will; ok is false, with nil err, when no such live client exists.
	DisconnectClient(clientID string, cleanSession, preventWill bool) (ok bool, err error)

	// SendToClientWithProperties behaves like SendToClient but also attaches
	// props as real MQTT5 packet properties (see PublishWithProperties).
	SendToClientWithProperties(
		clientID, topic string, payload []byte, qos byte, props MQTT5Properties,
	) (ok bool, err error)
}

MQTTPublisher publishes messages to the MQTT broker and can inspect or target the broker's individually connected clients.

type Provider 

type Provider struct{}

Provider implements service.Provider for the IoT Data Plane service.

func (*Provider) Init 

func (p *Provider) Init(ctx *service.AppContext) (service.Registerable, error)

Init initialises the IoT Data Plane backend and handler.

func (*Provider) Name 

func (p *Provider) Name() string

Name returns the service provider name.

type Resettable 

type Resettable interface {
	Reset()
}

Resettable is an optional interface a StorageBackend may implement to support full state reset.

type RetainedMessage 

type RetainedMessage struct {
	Topic   string
	Payload []byte
	// UserProperties holds the raw (base64-decoded) bytes of the MQTT5 user
	// properties JSON array supplied on the Publish call that established
	// this retained value, or nil when none were set. Mirrors
	// GetRetainedMessageOutput.UserProperties in the real SDK.
	UserProperties   []byte
	Qos              int32
	LastModifiedTime int64 // epoch milliseconds
}

RetainedMessage holds the details of a retained MQTT message stored by IoT.

type SessionInfo 

type SessionInfo struct {
	RemoteAddr    string
	LocalAddr     string
	SessionExpiry uint32
	KeepAlive     uint16
	Clean         bool
	ExpiryKnown   bool
}

SessionInfo describes a live broker session as GetConnection reports it. ExpiryKnown is false when the client stated no expiry (MQTT 3.1.1 never does).

type Snapshottable 

type Snapshottable interface {
	Snapshot(ctx context.Context) []byte
	Restore(context.Context, []byte) error
}

Snapshottable is an optional interface a StorageBackend may implement to support snapshot/restore for persistence or test isolation.

type StorageBackend 

type StorageBackend interface {
	Publish(topic string, payload []byte, qos int32, retain bool, props MQTT5Properties) error
	SetBroker(broker MQTTPublisher)
	GetThingShadow(thingName, shadowName string) ([]byte, error)
	UpdateThingShadow(thingName, shadowName string, document []byte) ([]byte, error)
	DeleteThingShadow(thingName, shadowName string) ([]byte, error)
	ListNamedShadowsForThing(thingName string) ([]string, error)
	ListThingsWithShadows() []string
	RegisterConnection(clientID, sourceIP string) error
	DeleteConnection(clientID string) error
	DeleteConnectionWithOptions(clientID string, cleanSession, preventWill bool) error
	ListConnections() []*Connection
	GetConnection(clientID string) (*Connection, error)
	ListSubscriptions(clientID string) ([]SubscriptionSummary, error)
	SendDirectMessage(
		clientID, topic string,
		payload []byte,
		qos int32,
		props MQTT5Properties,
	) error
	StoreRetainedMessage(topic string, payload []byte, qos int32, userProperties []byte) error
	GetRetainedMessage(topic string) (*RetainedMessage, error)
	ListRetainedMessages() ([]*RetainedMessage, error)
	Reset()
}

StorageBackend defines the interface for the IoT Data Plane backend.

type SubscriptionSummary added in v1.2.0 

type SubscriptionSummary struct {
	TopicFilter string
	QoS         byte
}

SubscriptionSummary is a single topic-filter/QoS pair describing one of a client's live MQTT subscriptions, mirroring the real SDK's types.SubscriptionSummary (topicFilter, qos).

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.