iotdataplane

package
v1.2.0 Latest Latest
Warning

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

 Go to latest Published: Aug 3, 2026 License: MIT Imports: 23 Imported by: 0

README

IoT Data Plane

Parity grade: A- · SDK aws-sdk-go-v2/service/iotdataplane@v1.35.0 · last audited 2026-07-25 (058bf0373)

Coverage

Metric Value
Operations audited 11 (8 ok, 2 partial, 1 gap)
Known gaps 5
Deferred items 1
Resource leaks clean
Known gaps
  • Publish with no MQTT broker wired logs a warning and silently drops the message (ErrNoBroker path in backend.go Publish()). This is intentional degradation, not a disguised no-op -- when a broker IS wired (see cli.go startup, out of scope for this service-only pass) the message is delivered for real, retain/qos forwarded. Additionally, the MQTTPublisher interface (services/iotdataplane/interfaces.go) only carries topic/payload/retain/qos -- contentType/correlationData/messageExpiry/payloadFormatIndicator/responseTopic are parsed and validated at the HTTP layer but never reach live MQTT subscribers, since forwarding them would require extending MQTTPublisher and its only real implementation (services/iot/broker.go, backed by mochi-mqtt), which is outside this service's own scope. No AWS-modeled response surface within iotdataplane echoes these fields back (GetRetainedMessageOutput only carries userProperties, which IS wired through), so this has no other observable wire-parity impact. No further work identified without cross-service broker changes.
  • UnsupportedDocumentEncodingException (real AWS error, modeled for GetThingShadow/DeleteThingShadow/UpdateThingShadow) is never returned -- no Content-Encoding-based validation exists. Left unimplemented: re-verified this pass via targeted web search (AWS API reference, boto3 docs) and still found no documented trigger condition (e.g. which Content-Encoding values are rejected, or whether it's Accept-Encoding-driven). Speculative validation risks a wrong-shape fix. Candidate for a future audit pass with real-AWS verification first (e.g. a live AWS account probe).
  • ListSubscriptions always returns an empty subscriptions array, even for a tracked/connected client. Real per-client subscription state DOES exist elsewhere in the repo -- the mochi-mqtt broker (services/iot/broker.go, github.com/mochi-mqtt/server/v2) tracks each client's live subscriptions in cl.State.Subscriptions -- but it is not reachable from this package: the MQTTPublisher interface (interfaces.go) this backend depends on only exposes topic-broadcast Publish(), and extending it to expose subscription queries would require changing services/iot/broker.go (out of scope for this pass; that directory was explicitly off-limits). Returning an honestly empty list for a genuinely-tracked client was chosen over fabricating topic filters. Candidate follow-up: add a ListSubscriptions(clientID) method to MQTTPublisher backed by Broker.server.Load().Clients.Get(clientID).State.Subscriptions, then have InMemoryBackend.ListSubscriptions call through it when a broker is wired.
  • SendDirectMessage delivers by broadcasting on the target topic through the same broker-backed path as Publish, not by sending directly to the named client the way real AWS does. Real SendDirectMessage explicitly does not require the receiving client to be subscribed to the topic ("the receiving client does not need to subscribe to the topic"); gopherstack's only broker primitive (MQTTPublisher.Publish, backed by mochi-mqtt's s.Publish) has no per-client-addressed send, so a client that isn't subscribed to the given topic will NOT observe a SendDirectMessage the way it would against real AWS. This is a deliberate, documented choice (wiring into the real delivery path so at least topic-subscribers observe it, rather than writing to a dead-end store no caller could ever observe) -- see InMemoryBackend.SendDirectMessage's doc comment. Fixing this for real would need mochi-mqtt's client-targeted write path (s.Clients.Get(clientID) + a raw PUBLISH write), which lives in services/iot/broker.go, out of scope here. confirmation/timeout (real AWS: wait for a QoS-1 PUBACK, HTTP 504 on timeout) select QoS 0-vs-1 on the outgoing publish but never actually block or time out, since MQTTPublisher.Publish is synchronous/fire-and-forget with no ack channel.
  • GetConnection omits cleanSession/disconnectReason/disconnectedSince/keepAliveDuration/sessionExpiry/sourcePort/targetIp/targetPort/thingName/vpcEndpointId from its response for every client, tracked or not -- gopherstack's connections table (populated only by the gopherstack-only RegisterConnection admin extension) never had this data to begin with (no real MQTT CONNECT packet is parsed anywhere in this service). Omitted (not zero-valued) so a real SDK client decodes these exactly as if the server had never observed them, which is wire-compatible even though it under-reports what a live AWS endpoint would return.
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 exceeds the maximum allowed size. Wire error code "RequestEntityTooLargeException" (real AWS iotdataplane exception, modeled only for UpdateThingShadow).

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 {
	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
}

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) 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) 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) 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 by delegating to the backend if it supports it.

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 by delegating to the backend if it supports it.

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) 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) error

Publish delivers a message to the given MQTT topic. 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) 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 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

	// 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)
}

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 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) 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
	ListConnections() []*Connection
	GetConnection(clientID string) (*Connection, error)
	ListSubscriptions(clientID string) ([]SubscriptionSummary, error)
	SendDirectMessage(clientID, topic string, payload []byte, qos int32) 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.