README
¶
IoT Data Plane
Parity grade: A · SDK aws-sdk-go-v2/service/iotdataplane@v1.32.20 · last audited 2026-07-13 (57398ee1)
Coverage
| Operations audited | 8 (8 ok) |
| Known gaps | 3 |
| 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. No further work identified without broker wiring changes, which live outside services/iotdataplane/.
- UnsupportedDocumentEncodingException (real AWS error, modeled for GetThingShadow/DeleteThingShadow/UpdateThingShadow) is never returned -- no Content-Encoding-based validation exists. Left unimplemented: no clear trigger condition was verified against real AWS behavior, and speculative validation risks a wrong-shape fix. Candidate for a future audit pass with real-AWS verification first.
- maxShadowsPerThing=100 (backend.go) is a soft self-imposed cap, not verified against an authoritative AWS quota number -- left unchanged this pass (low confidence either way, non-blocking).
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 ¶
- Variables
- type Connection
- type Handler
- func (h *Handler) ChaosOperations() []string
- func (h *Handler) ChaosRegions() []string
- func (h *Handler) ChaosServiceName() string
- func (h *Handler) ExtractOperation(c *echo.Context) string
- func (h *Handler) ExtractResource(c *echo.Context) string
- func (h *Handler) GetSupportedOperations() []string
- func (h *Handler) Handler() echo.HandlerFunc
- func (h *Handler) MatchPriority() int
- func (h *Handler) Name() string
- func (h *Handler) Reset()
- func (h *Handler) Restore(ctx context.Context, data []byte) error
- func (h *Handler) RouteMatcher() service.Matcher
- func (h *Handler) Snapshot(ctx context.Context) []byte
- type InMemoryBackend
- func (b *InMemoryBackend) AddConnectionInternal(clientID string)
- func (b *InMemoryBackend) AddShadowInternal(thingName, shadowName string, document []byte)
- func (b *InMemoryBackend) DeleteConnection(clientID string) error
- func (b *InMemoryBackend) DeleteThingShadow(thingName, shadowName string) ([]byte, error)
- func (b *InMemoryBackend) GetRetainedMessage(topic string) (*RetainedMessage, error)
- func (b *InMemoryBackend) GetThingShadow(thingName, shadowName string) ([]byte, error)
- func (b *InMemoryBackend) ListConnections() []*Connection
- func (b *InMemoryBackend) ListNamedShadowsForThing(thingName string) ([]string, error)
- func (b *InMemoryBackend) ListRetainedMessages() ([]*RetainedMessage, error)
- func (b *InMemoryBackend) ListThingsWithShadows() []string
- func (b *InMemoryBackend) Publish(topic string, payload []byte, qos int32, retain bool) error
- func (b *InMemoryBackend) RegisterConnection(clientID, sourceIP string) error
- func (b *InMemoryBackend) Reset()
- func (b *InMemoryBackend) Restore(ctx context.Context, data []byte) error
- func (b *InMemoryBackend) SetBroker(broker MQTTPublisher)
- func (b *InMemoryBackend) Snapshot(ctx context.Context) []byte
- func (b *InMemoryBackend) StoreRetainedMessage(topic string, payload []byte, qos int32) error
- func (b *InMemoryBackend) UpdateThingShadow(thingName, shadowName string, document []byte) ([]byte, error)
- type MQTTPublisher
- type Provider
- type Resettable
- type RetainedMessage
- type Snapshottable
- type StorageBackend
Constants ¶
This section is empty.
Variables ¶
var ErrConnectionExists = errors.New("connection already exists")
ErrConnectionExists is returned when trying to register a clientID that is already connected.
var ErrNilAppContext = errors.New("AppContext is required")
ErrNilAppContext is returned when Provider.Init is called with a nil AppContext.
var ErrNoBroker = errors.New("no mqtt broker configured")
ErrNoBroker is returned when no MQTT broker has been wired.
var ErrNoSnapshot = errors.New("backend does not support restore")
ErrNoSnapshot is returned when a backend does not support snapshot/restore.
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).
var ErrRetainedMessageNotFound = errors.New("retained message not found")
ErrRetainedMessageNotFound is returned when no retained message exists for a topic.
var ErrShadowNotFound = errors.New("shadow not found")
ErrShadowNotFound is returned when a thing shadow is not found.
var ErrValidation = errors.New("InvalidRequestException")
ErrValidation is returned for invalid input parameters.
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 ¶
ChaosOperations returns all operations that can be fault-injected.
func (*Handler) ChaosRegions ¶
ChaosRegions returns all regions this IoT Data Plane instance handles.
func (*Handler) ChaosServiceName ¶
ChaosServiceName returns the lowercase AWS service name for fault rule matching.
func (*Handler) ExtractOperation ¶
ExtractOperation returns the operation name.
func (*Handler) ExtractResource ¶
ExtractResource extracts the topic or thing name from the URL path.
func (*Handler) GetSupportedOperations ¶
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 ¶
MatchPriority returns the routing priority for the IoT Data Plane handler.
func (*Handler) Reset ¶
func (h *Handler) Reset()
Reset clears all handler state by delegating to the backend.
func (*Handler) Restore ¶
Restore implements persistence by delegating to the backend if it supports it.
func (*Handler) RouteMatcher ¶
RouteMatcher returns a function matching IoT Data Plane requests.
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 removes an MQTT client connection from the backend. If the clientID does not exist the operation is a no-op (idempotent). 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) 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) ListThingsWithShadows ¶
func (b *InMemoryBackend) ListThingsWithShadows() []string
ListThingsWithShadows returns the sorted list of thing names that have at least one shadow.
func (*InMemoryBackend) Publish ¶
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) 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) error
StoreRetainedMessage saves a retained MQTT message for the given topic. Calling this with an empty payload removes the retained message for that topic. 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 ¶
MQTTPublisher publishes a message to an MQTT topic.
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.
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
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
StoreRetainedMessage(topic string, payload []byte, qos int32) error
GetRetainedMessage(topic string) (*RetainedMessage, error)
ListRetainedMessages() ([]*RetainedMessage, error)
Reset()
}
StorageBackend defines the interface for the IoT Data Plane backend.
Source Files