im

package
v0.0.0-...-02ded7e Latest Latest
Warning

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

 Go to latest Published: Oct 24, 2026 License: MIT Imports: 18 Imported by: 0

README

im

Interaction Model engine for Matter (Spec Chapter 8, 10).

Architecture

                    ┌──────────────────────────────────────────┐
                    │                Engine                     │
                    │                                          │
                    │   OnMessage(header, payload)             │
                    │          │                               │
                    │          ▼                               │
                    │   ┌──────────────────────────────┐       │
                    │   │   Opcode Router              │       │
                    │   │   0x02 → ReadHandler         │       │
                    │   │   0x06 → WriteHandler        │       │
                    │   │   0x08 → InvokeHandler       │       │
                    │   │   0x01 → StatusResponse      │       │
                    │   └──────────────────────────────┘       │
                    │          │                               │
                    │          ▼                               │
                    │   ┌──────────────────────────────┐       │
                    │   │   Dispatcher                 │       │
                    │   │   Routes to cluster impls    │       │
                    │   └──────────────────────────────┘       │
                    └──────────────────────────────────────────┘
                                       │
                    ┌──────────────────┼──────────────────┐
                    ▼                  ▼                  ▼
             pkg/datamodel       pkg/acl           pkg/exchange
             (Cluster impls)     (ACL checks)      (ExchangeContext)

Usage

Create Engine
engine := im.NewEngine(im.EngineConfig{
    Dispatcher: myDispatcher,
    ACLChecker: aclChecker,  // optional
    MaxPayload: 1280,        // optional, defaults to 1232
})
Register with Exchange Manager
exchangeManager.RegisterDelegate(im.ProtocolID, engine)
Implement Dispatcher
type MyDispatcher struct {
    router *datamodel.Router
}

func (d *MyDispatcher) ReadAttribute(ctx context.Context, req *im.AttributeReadRequest, w *tlv.Writer) error {
    cluster, err := d.router.GetCluster(req.Path.Endpoint, req.Path.Cluster)
    if err != nil {
        return im.ErrClusterNotFound
    }
    return cluster.ReadAttribute(ctx, req.ToDataModelRequest(), w)
}

func (d *MyDispatcher) WriteAttribute(ctx context.Context, req *im.AttributeWriteRequest, r *tlv.Reader) error {
    // Similar pattern...
}

func (d *MyDispatcher) InvokeCommand(ctx context.Context, req *im.CommandInvokeRequest, r *tlv.Reader) ([]byte, error) {
    // Similar pattern...
}

Message Flow

Read:
  C ── ReadRequest (0x02) ──────▶ S
  C ◀── ReportData (0x05) ─────── S
  C ── StatusResponse (0x01) ───▶ S  (if chunked)

Write:
  C ── WriteRequest (0x06) ─────▶ S
  C ◀── WriteResponse (0x07) ──── S

Invoke:
  C ── InvokeRequest (0x08) ────▶ S
  C ◀── InvokeResponse (0x09) ─── S

Handlers

Handler Opcode State Machine
ReadHandler 0x02 Idle → Processing → SendingReport → Idle
WriteHandler 0x06 Idle → Processing → Idle
InvokeHandler 0x08 Idle → Processing → SendingResponse → Idle

Error Mapping

Error IM Status
ErrClusterNotFound UnsupportedCluster (0xC3)
ErrAttributeNotFound UnsupportedAttribute (0x86)
ErrCommandNotFound UnsupportedCommand (0x81)
ErrAccessDenied UnsupportedAccess (0x7E)
ErrConstraintError ConstraintError (0x87)

Chunking

Large payloads are split across multiple messages:

// Assembler: receive chunked request
assembler := im.NewAssembler(im.ChunkTypeInvokeRequest)
for msg := range incomingMessages {
    complete, err := assembler.AddChunk(msg)
    if complete {
        fullPayload := assembler.Data()
        // Process complete request
    }
}

// Fragmenter: send chunked response
fragmenter := im.NewFragmenter(im.ChunkTypeInvokeResponse, maxPayload)
chunks := fragmenter.Fragment(largeResponse)
for _, chunk := range chunks {
    send(chunk)
    waitForStatusResponse()
}

Events

// Publish event
mgr := im.NewEventManager(im.EventManagerConfig{MaxEvents: 100})
mgr.PublishEvent(im.EventRecord{
    Path:        im.EventPath{Endpoint: 0, Cluster: 0x0028, Event: 0},
    EventNumber: mgr.NextEventNumber(),
    Priority:    im.EventPriorityCritical,
    Timestamp:   time.Now(),
    Data:        eventData,
})

// Build unsolicited report
reporter := im.NewEventReporter(mgr)
report := reporter.BuildUnsolicitedReport(fabricIndex, []im.EventPath{...})

Test Infrastructure

SecureTestIMPair

Two connected IM engines with encrypted sessions for E2E testing.

  Client (0)                          Server (1)
  ──────────                          ──────────
  im.Client                           im.Engine
      │                                   │
      ▼                                   ▼
  exchange.Manager ◀──── Pipe ────▶ exchange.Manager
      │                                   │
      ▼                                   ▼
  SecureContext(1,2)               SecureContext(2,1)

pair, _ := im.NewSecureTestIMPair(im.SecureTestIMPairConfig{
    Dispatchers: [2]im.Dispatcher{nil, mockDispatcher},
})
defer pair.Close()

// Client sends InvokeRequest to Server
result, err := pair.Client(0).InvokeWithStatus(
    ctx,
    pair.Session(0),     // Client's secure session
    pair.PeerAddress(1), // Server's address
    1, 0x0006, 0x01,     // Endpoint, Cluster, Command
    nil,
)

// Verify server received
calls := mockDispatcher.InvokeCalls()
MockDispatcher

Records IM operations for test verification.

mock := im.NewMockDispatcher()

// Configure responses
mock.SetInvokeResult(responseData, nil)
mock.SetReadResult(true, nil)

// After test
calls := mock.InvokeCalls()   // []InvokeCall
reads := mock.ReadCalls()     // []ReadCall
mock.Reset()
Bidirectional
pair, _ := im.NewSecureTestIMPair(im.SecureTestIMPairConfig{
    Dispatchers: [2]im.Dispatcher{mockClient, mockServer},
})

// Client → Server
pair.Client(0).InvokeWithStatus(ctx, pair.Session(0), pair.PeerAddress(1), ...)

// Server → Client
pair.Client(1).InvokeWithStatus(ctx, pair.Session(1), pair.PeerAddress(0), ...)

Documentation

Overview

Package im implements the Matter Interaction Model.

Index

Constants

View Source 
const (
	// DefaultMTU is the default IPv6 minimum MTU.
	DefaultMTU = 1280

	// MessageHeaderOverhead is approximate header overhead.
	// This includes message header, protocol header, and encryption overhead.
	MessageHeaderOverhead = 100

	// DefaultMaxPayload is the default maximum payload size per chunk.
	DefaultMaxPayload = DefaultMTU - MessageHeaderOverhead
)

Default MTU values per Matter spec.

View Source 
const DefaultRequestTimeout = 30 * time.Second

DefaultRequestTimeout is the default timeout for IM requests.

View Source 
const ProtocolID message.ProtocolID = 0x0001

ProtocolID is the Interaction Model protocol ID. Spec: Section 10.2.1

Variables

View Source 
var (
	ErrChunkingNotInProgress = errors.New("chunking: no assembly in progress")
	ErrChunkingInProgress    = errors.New("chunking: assembly already in progress")
	ErrChunkingPayloadEmpty  = errors.New("chunking: empty payload")
	ErrChunkingMTUTooSmall   = errors.New("chunking: MTU too small for header")
)

Chunking-related errors.

View Source 
var (
	ErrClientTimeout      = errors.New("im: request timeout")
	ErrClientClosed       = errors.New("im: client closed")
	ErrUnexpectedResponse = errors.New("im: unexpected response type")
	ErrCommandFailed      = errors.New("im: command failed")
)

Client errors.

View Source 
var (
	// ErrClusterNotFound indicates the cluster doesn't exist on the endpoint.
	ErrClusterNotFound = errors.New("im: cluster not found")

	// ErrAttributeNotFound indicates the attribute doesn't exist on the cluster.
	ErrAttributeNotFound = errors.New("im: attribute not found")

	// ErrCommandNotFound indicates the command doesn't exist on the cluster.
	ErrCommandNotFound = errors.New("im: command not found")

	// ErrAccessDenied indicates ACL check failed.
	ErrAccessDenied = errors.New("im: access denied")

	// ErrUnsupportedWrite indicates the attribute doesn't support writes.
	ErrUnsupportedWrite = errors.New("im: unsupported write")

	// ErrUnsupportedRead indicates the attribute doesn't support reads.
	ErrUnsupportedRead = errors.New("im: unsupported read")

	// ErrConstraintError indicates a constraint violation (e.g., invalid value).
	ErrConstraintError = errors.New("im: constraint error")

	// ErrDataVersionMismatch indicates optimistic lock failure.
	ErrDataVersionMismatch = errors.New("im: data version mismatch")

	// ErrNeedsTimedInteraction indicates the operation requires a timed interaction.
	ErrNeedsTimedInteraction = errors.New("im: needs timed interaction")

	// ErrInvalidPath indicates the path is malformed or has wildcard when not allowed.
	ErrInvalidPath = errors.New("im: invalid path")

	// ErrBusy indicates the engine is busy and cannot process the request.
	ErrBusy = errors.New("im: busy")

	// ErrResourceExhausted indicates resource limits exceeded.
	ErrResourceExhausted = errors.New("im: resource exhausted")
)

IM engine errors.

View Source 
var (
	ErrInvokeHandlerBusy     = errors.New("invoke handler: busy processing another request")
	ErrInvokeTimedMismatch   = errors.New("invoke handler: timed request mismatch")
	ErrInvokeCommandNotFound = errors.New("invoke handler: command not found")
	ErrInvokeInvalidPath     = errors.New("invoke handler: invalid command path")
)

InvokeHandler errors.

View Source 
var (
	ErrReadHandlerBusy  = errors.New("read handler: busy processing another request")
	ErrReadPathNotFound = errors.New("read handler: path not found")
	ErrReadAccessDenied = errors.New("read handler: access denied")
)

ReadHandler errors.

View Source 
var (
	ErrWriteHandlerBusy   = errors.New("write handler: busy processing another request")
	ErrWriteTimedMismatch = errors.New("write handler: timed request mismatch")
	ErrWriteWildcardPath  = errors.New("write handler: wildcard paths not supported")
	ErrWriteListOperation = errors.New("write handler: list operations not supported")
)

WriteHandler errors.

Functions

func DecodeInvokeRequest 

func DecodeInvokeRequest(data []byte) (*message.InvokeRequestMessage, error)

DecodeInvokeRequest decodes an invoke request message.

func DecodeInvokeResponse 

func DecodeInvokeResponse(data []byte) (*imsg.InvokeResponseMessage, error)

DecodeInvokeResponse decodes an InvokeResponseMessage from TLV.

func DecodeReadRequest 

func DecodeReadRequest(data []byte) (*message.ReadRequestMessage, error)

DecodeReadRequest decodes a read request message.

func DecodeReportData 

func DecodeReportData(data []byte) (*imsg.ReportDataMessage, error)

DecodeReportData decodes a ReportDataMessage from TLV.

func DecodeStatusResponse 

func DecodeStatusResponse(data []byte) (*message.StatusResponseMessage, error)

DecodeStatusResponse decodes a status response message.

func DecodeWriteRequest 

func DecodeWriteRequest(data []byte) (*message.WriteRequestMessage, error)

DecodeWriteRequest decodes a write request message.

func EncodeInvokeRequest 

func EncodeInvokeRequest(req *imsg.InvokeRequestMessage) ([]byte, error)

EncodeInvokeRequest encodes an InvokeRequestMessage to TLV.

func EncodeInvokeResponse 

func EncodeInvokeResponse(msg *message.InvokeResponseMessage) ([]byte, error)

EncodeInvokeResponse encodes an invoke response message.

func EncodeMessage 

func EncodeMessage(encode func(w *tlv.Writer) error) ([]byte, error)

EncodeMessage encodes a message to TLV bytes. This is a helper for estimating actual encoded size.

func EncodeReadRequest 

func EncodeReadRequest(req *imsg.ReadRequestMessage) ([]byte, error)

EncodeReadRequest encodes a ReadRequestMessage to TLV.

func EncodeReportData 

func EncodeReportData(msg *message.ReportDataMessage) ([]byte, error)

EncodeReportData encodes a report data message.

func EncodeStatusResponse 

func EncodeStatusResponse(status message.Status) ([]byte, error)

EncodeStatusResponse encodes a status response message.

func EncodeWriteResponse 

func EncodeWriteResponse(msg *message.WriteResponseMessage) ([]byte, error)

EncodeWriteResponse encodes a write response message.

func ErrorToStatus 

func ErrorToStatus(err error) message.Status

ErrorToStatus maps an error to an IM status code. This follows the Matter spec mapping of errors to status codes.

func StatusToError 

func StatusToError(status message.Status) error

StatusToError maps an IM status code to an error.

Types

type Assembler 

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

Assembler collects chunked messages until complete. It accumulates the array elements from each chunk until MoreChunkedMessages is false, then returns the complete message.

Per Spec 10.3.2: When a message is chunked, each chunk contains partial array elements. The receiver accumulates these until the final chunk (MoreChunkedMessages=false).

func NewAssembler 

func NewAssembler() *Assembler

NewAssembler creates a new chunking assembler.

func (*Assembler) AddInvokeResponse 

func (a *Assembler) AddInvokeResponse(msg *message.InvokeResponseMessage) (*message.InvokeResponseMessage, bool, error)

AddInvokeResponse adds an InvokeResponseMessage chunk. Returns (completeMessage, isComplete, error).

func (*Assembler) AddReportData 

func (a *Assembler) AddReportData(msg *message.ReportDataMessage) (*message.ReportDataMessage, bool, error)

AddReportData adds a ReportDataMessage chunk. Returns (completeMessage, isComplete, error).

func (*Assembler) AddWriteRequest 

func (a *Assembler) AddWriteRequest(msg *message.WriteRequestMessage) (*message.WriteRequestMessage, bool, error)

AddWriteRequest adds a WriteRequestMessage chunk. Returns (completeMessage, isComplete, error). If isComplete is true, completeMessage contains the assembled message.

func (*Assembler) ChunkType 

func (a *Assembler) ChunkType() ChunkType

ChunkType returns the current chunk type being assembled.

func (*Assembler) IsAssembling 

func (a *Assembler) IsAssembling() bool

IsAssembling returns true if assembly is in progress.

func (*Assembler) Reset 

func (a *Assembler) Reset()

Reset clears the assembler state.

type AttributeReadRequest 

type AttributeReadRequest struct {
	// Path identifies the attribute to read.
	Path message.AttributePathIB

	// IMContext is the IM request context.
	IMContext *RequestContext

	// IsFabricFiltered indicates fabric-filtered read.
	IsFabricFiltered bool
}

AttributeReadRequest contains parameters for reading an attribute via IM.

func (*AttributeReadRequest) ToDataModelRequest 

func (r *AttributeReadRequest) ToDataModelRequest() datamodel.ReadAttributeRequest

ToDataModelRequest converts to a datamodel.ReadAttributeRequest.

type AttributeReader 

type AttributeReader func(
	ctx *ReadContext,
	path message.AttributePathIB,
) (*AttributeResult, error)

AttributeReader is called to read attribute data. It receives the attribute path and returns the TLV-encoded data.

type AttributeResult 

type AttributeResult struct {
	// DataVersion is the current data version of the cluster.
	DataVersion message.DataVersion

	// Data is the TLV-encoded attribute value.
	Data []byte

	// Status is set if the read failed with a status.
	Status *message.StatusIB
}

AttributeResult is the result of reading an attribute.

type AttributeWriteRequest 

type AttributeWriteRequest struct {
	// Path identifies the attribute to write.
	Path message.AttributePathIB

	// IMContext is the IM request context.
	IMContext *RequestContext

	// IsTimed indicates this is a timed write.
	IsTimed bool

	// DataVersion is the expected data version (nil = no check).
	DataVersion *message.DataVersion
}

AttributeWriteRequest contains parameters for writing an attribute via IM.

func (*AttributeWriteRequest) ToDataModelRequest 

func (r *AttributeWriteRequest) ToDataModelRequest() datamodel.WriteAttributeRequest

ToDataModelRequest converts to a datamodel.WriteAttributeRequest.

type ChunkType 

type ChunkType int

ChunkType identifies what type of message is being chunked. 

const (
	ChunkTypeNone ChunkType = iota
	ChunkTypeWriteRequest
	ChunkTypeReportData
	ChunkTypeInvokeResponse
)

func (ChunkType) String 

func (t ChunkType) String() string

String returns a human-readable name for the chunk type.

type Client 

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

Client provides client-side IM operations for sending cluster commands. It wraps the exchange layer to provide a synchronous request-response API.

Usage: 

client := im.NewClient(exchangeManager)
resp, err := client.Invoke(ctx, session, path, request)

func NewClient 

func NewClient(config ClientConfig) *Client

NewClient creates a new IM client.

func (*Client) InvokeRequest 

func (c *Client) InvokeRequest(
	ctx context.Context,
	sess *session.SecureContext,
	peerAddr transport.PeerAddress,
	endpointID uint16,
	clusterID uint32,
	commandID uint32,
	requestData []byte,
) ([]byte, error)

InvokeRequest sends a command to a cluster and waits for the response.

Parameters:

  • ctx: Context for cancellation and timeout
  • sess: Secure session to use for the request
  • peerAddr: Peer network address
  • endpointID: Target endpoint (typically 0 for root)
  • clusterID: Target cluster ID
  • commandID: Command ID to invoke
  • requestData: TLV-encoded request fields (can be nil for commands with no fields)

Returns the TLV-encoded response data and error.

func (*Client) InvokeWithStatus 

func (c *Client) InvokeWithStatus(
	ctx context.Context,
	sess *session.SecureContext,
	peerAddr transport.PeerAddress,
	endpointID uint16,
	clusterID uint32,
	commandID uint32,
	requestData []byte,
) (*InvokeResult, error)

InvokeWithStatus sends a command and returns the full result including status.

func (*Client) ReadAttribute 

func (c *Client) ReadAttribute(
	ctx context.Context,
	sess *session.SecureContext,
	peerAddr transport.PeerAddress,
	endpointID uint16,
	clusterID uint32,
	attributeID uint32,
) ([]byte, error)

ReadAttribute reads a single attribute from a cluster.

type ClientConfig 

type ClientConfig struct {
	// ExchangeManager handles message exchanges.
	// Required.
	ExchangeManager *exchange.Manager

	// Timeout for requests. Defaults to DefaultRequestTimeout if zero.
	Timeout time.Duration

	// LoggerFactory creates loggers for the client.
	// If nil, logging is disabled.
	LoggerFactory logging.LoggerFactory
}

ClientConfig configures the Client.

type ClusterDispatcher 

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

ClusterDispatcher routes IM operations to registered clusters. This provides a simple data model for testing.

func NewClusterDispatcher 

func NewClusterDispatcher() *ClusterDispatcher

NewClusterDispatcher creates a new cluster dispatcher.

func (*ClusterDispatcher) InvokeCommand 

func (d *ClusterDispatcher) InvokeCommand(ctx context.Context, req *CommandInvokeRequest, r *tlv.Reader) ([]byte, error)

InvokeCommand implements Dispatcher.

func (*ClusterDispatcher) ReadAttribute 

func (d *ClusterDispatcher) ReadAttribute(ctx context.Context, req *AttributeReadRequest, w *tlv.Writer) error

ReadAttribute implements Dispatcher.

func (*ClusterDispatcher) RegisterCluster 

func (d *ClusterDispatcher) RegisterCluster(endpoint datamodel.EndpointID, cluster datamodel.Cluster)

RegisterCluster registers a cluster at an endpoint.

func (*ClusterDispatcher) WriteAttribute 

func (d *ClusterDispatcher) WriteAttribute(ctx context.Context, req *AttributeWriteRequest, r *tlv.Reader) error

WriteAttribute implements Dispatcher.

type CommandHandler 

type CommandHandler func(
	ctx *InvokeContext,
	path message.CommandPathIB,
	fields []byte,
) (*CommandResult, error)

CommandHandler is called to process an invoke request. It receives the command path and raw TLV command fields, and returns response data (raw TLV) or an error status.

type CommandInvokeRequest 

type CommandInvokeRequest struct {
	// Path identifies the command to invoke.
	Path message.CommandPathIB

	// IMContext is the IM request context.
	IMContext *RequestContext

	// IsTimed indicates this is a timed invoke.
	IsTimed bool
}

CommandInvokeRequest contains parameters for invoking a command via IM.

func (*CommandInvokeRequest) ToDataModelRequest 

func (r *CommandInvokeRequest) ToDataModelRequest() datamodel.InvokeRequest

ToDataModelRequest converts to a datamodel.InvokeRequest.

type CommandResult 

type CommandResult struct {
	// ResponsePath is the command path for the response.
	// Typically the same as the request path for server commands.
	ResponsePath message.CommandPathIB

	// ResponseData is the TLV-encoded response data.
	// nil if command has no response data.
	ResponseData []byte

	// Status is set if the command failed with a status instead of response.
	Status *message.StatusIB
}

CommandResult is the result of a command invocation.

type Dispatcher 

type Dispatcher interface {
	// ReadAttribute reads an attribute value.
	// Returns error if the cluster/attribute doesn't exist or access denied.
	ReadAttribute(ctx context.Context, req *AttributeReadRequest, w *tlv.Writer) error

	// WriteAttribute writes an attribute value.
	// Returns error if the cluster/attribute doesn't exist, access denied, or constraint violation.
	WriteAttribute(ctx context.Context, req *AttributeWriteRequest, r *tlv.Reader) error

	// InvokeCommand invokes a cluster command.
	// Returns response TLV data (may be nil) and error.
	InvokeCommand(ctx context.Context, req *CommandInvokeRequest, r *tlv.Reader) ([]byte, error)
}

Dispatcher routes IM operations to cluster implementations. This is the bridge between the IM engine and the data model.

type Engine 

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

Engine is the Interaction Model engine. It implements exchange.ExchangeDelegate for the IM protocol.

This simplified engine supports:

  • ReadRequest → ReportData
  • WriteRequest → WriteResponse
  • InvokeRequest → InvokeResponse
  • StatusResponse (for chunked flows)

It does NOT support (for commissioning simplicity):

  • Subscriptions
  • Timed interactions
  • Complex chunking

Spec Reference: Chapter 8 "Interaction Model Specification" C++ Reference: src/app/InteractionModelEngine.cpp

func NewEngine 

func NewEngine(config EngineConfig) *Engine

NewEngine creates a new IM engine.

func (*Engine) GetProtocolID 

func (e *Engine) GetProtocolID() message.ProtocolID

GetProtocolID returns the protocol ID for registration with ExchangeManager.

func (*Engine) OnClose 

func (e *Engine) OnClose(ctx *exchange.ExchangeContext)

OnClose implements exchange.ExchangeDelegate.

func (*Engine) OnMessage 

func (e *Engine) OnMessage(
	ctx *exchange.ExchangeContext,
	header *message.ProtocolHeader,
	payload []byte,
) ([]byte, error)

OnMessage implements exchange.ExchangeDelegate. This is the main entry point for IM messages.

The engine sends responses directly via ctx.SendMessage with the correct response opcode (matching the C++ SDK architecture), then returns (nil, nil) so the exchange layer doesn't send again.

Spec: 8.2.4 "Action" - defines valid opcodes C++ Reference: InteractionModelEngine::OnMessageReceived

type EngineConfig 

type EngineConfig struct {
	// Dispatcher routes operations to cluster implementations.
	// Required.
	Dispatcher Dispatcher

	// ACLChecker performs access control checks.
	// Optional - if nil, ACL checks are skipped.
	ACLChecker *acl.Checker

	// MaxPayload is the maximum payload size for responses.
	// Defaults to DefaultMaxPayload if 0.
	MaxPayload int

	// LoggerFactory is the factory for creating loggers.
	// If nil, logging is disabled.
	LoggerFactory logging.LoggerFactory
}

EngineConfig configures the Engine.

type EventListener 

type EventListener interface {
	// OnEvent is called when a new event is recorded.
	OnEvent(record *EventRecord)
}

EventListener is notified when events are generated.

type EventManager 

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

EventManager manages event generation and storage. It maintains a circular buffer of recent events per priority level and generates monotonically increasing event numbers.

func NewEventManager 

func NewEventManager(config EventManagerConfig) *EventManager

NewEventManager creates a new event manager.

func (*EventManager) AddListener 

func (m *EventManager) AddListener(listener EventListener)

AddListener registers a listener for event notifications.

func (*EventManager) Clear 

func (m *EventManager) Clear()

Clear removes all stored events.

func (*EventManager) GetEvents 

func (m *EventManager) GetEvents(
	path *EventPath,
	minEventNumber *message.EventNumber,
	fabricIndex uint8,
	priorities []EventPriority,
) []*EventRecord

GetEvents returns events matching the filter criteria.

func (*EventManager) GetLatestEventNumber 

func (m *EventManager) GetLatestEventNumber() message.EventNumber

GetLatestEventNumber returns the most recent event number.

func (*EventManager) PublishEvent 

func (m *EventManager) PublishEvent(
	endpointID message.EndpointID,
	clusterID message.ClusterID,
	eventID message.EventID,
	priority EventPriority,
	data []byte,
) message.EventNumber

PublishEvent generates and stores a new event. Returns the assigned event number.

func (*EventManager) PublishEventWithFabric 

func (m *EventManager) PublishEventWithFabric(
	endpointID message.EndpointID,
	clusterID message.ClusterID,
	eventID message.EventID,
	priority EventPriority,
	data []byte,
	fabricIndex uint8,
) message.EventNumber

PublishEventWithFabric generates a fabric-scoped event.

func (*EventManager) RemoveListener 

func (m *EventManager) RemoveListener(listener EventListener)

RemoveListener unregisters a listener.

type EventManagerConfig 

type EventManagerConfig struct {
	// MaxEvents is the maximum number of events to retain.
	// Oldest events are evicted when this limit is exceeded.
	// Default: 100
	MaxEvents int

	// MaxEventsPerPriority limits events per priority level.
	// Default: 50 per level
	MaxEventsPerPriority int
}

EventManagerConfig configures the EventManager.

type EventManagerPublisher 

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

EventManagerPublisher adapts EventManager to implement datamodel.EventPublisher. It handles TLV encoding of event payloads centrally.

Payload types should implement TLVMarshaler. If the payload is nil, an empty event data is used. If the payload is already []byte, it's used directly (for backwards compatibility).

func NewEventManagerPublisher 

func NewEventManagerPublisher(em *EventManager) *EventManagerPublisher

NewEventManagerPublisher creates a new EventManagerPublisher.

func (*EventManagerPublisher) PublishEvent 

func (p *EventManagerPublisher) PublishEvent(
	endpoint datamodel.EndpointID,
	cluster datamodel.ClusterID,
	eventID datamodel.EventID,
	priority datamodel.EventPriority,
	data interface{},
	fabricIndex uint8,
) (datamodel.EventNumber, error)

PublishEvent implements datamodel.EventPublisher. The data parameter can be:

  • nil: empty event data
  • TLVMarshaler: will be encoded via MarshalTLV
  • []byte: used directly (backwards compatibility)

type EventPath 

type EventPath struct {
	EndpointID message.EndpointID
	ClusterID  message.ClusterID
	EventID    message.EventID
}

EventPath identifies an event source.

func (EventPath) ToEventPathIB 

func (p EventPath) ToEventPathIB() message.EventPathIB

ToEventPathIB converts to the wire format.

type EventPriority 

type EventPriority uint8

EventPriority matches the spec event priority levels. 

const (
	EventPriorityDebug    EventPriority = 0
	EventPriorityInfo     EventPriority = 1
	EventPriorityCritical EventPriority = 2
)

func (EventPriority) String 

func (p EventPriority) String() string

String returns the name of the priority level.

type EventRecord 

type EventRecord struct {
	// Path identifies the event source.
	Path EventPath

	// EventNumber is the monotonically increasing event counter.
	EventNumber message.EventNumber

	// Priority is the event priority level.
	Priority EventPriority

	// Timestamp when the event was generated.
	Timestamp time.Time

	// Data is the raw TLV-encoded event data.
	Data []byte

	// FabricIndex if the event is fabric-scoped (0 if not).
	FabricIndex uint8
}

EventRecord represents a stored event.

func (*EventRecord) ToEventDataIB 

func (r *EventRecord) ToEventDataIB() message.EventDataIB

ToEventDataIB converts an EventRecord to wire format.

func (*EventRecord) ToEventReportIB 

func (r *EventRecord) ToEventReportIB() message.EventReportIB

ToEventReportIB converts an EventRecord to a report IB.

type EventReporter 

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

EventReporter provides methods to build event report messages.

func NewEventReporter 

func NewEventReporter(em *EventManager) *EventReporter

NewEventReporter creates a new event reporter.

func (*EventReporter) BuildReportData 

func (r *EventReporter) BuildReportData(
	events []*EventRecord,
	subscriptionID *message.SubscriptionID,
	suppressResponse bool,
) *message.ReportDataMessage

BuildReportData creates a ReportDataMessage containing events.

func (*EventReporter) BuildUnsolicitedReport 

func (r *EventReporter) BuildUnsolicitedReport(
	events []*EventRecord,
) *message.ReportDataMessage

BuildUnsolicitedReport creates an unsolicited report for new events. This is used when events need to be pushed immediately (e.g., PushTransportBegin).

type Fragmenter 

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

Fragmenter splits large messages into chunks that fit within MTU.

func NewFragmenter 

func NewFragmenter(maxPayload int) *Fragmenter

NewFragmenter creates a new fragmenter with the given max payload size.

func (*Fragmenter) FragmentInvokeResponse 

func (f *Fragmenter) FragmentInvokeResponse(msg *message.InvokeResponseMessage) ([]*message.InvokeResponseMessage, error)

FragmentInvokeResponse splits an InvokeResponseMessage into chunks. Returns a slice of chunked messages ready for transmission.

func (*Fragmenter) FragmentReportData 

func (f *Fragmenter) FragmentReportData(msg *message.ReportDataMessage) ([]*message.ReportDataMessage, error)

FragmentReportData splits a ReportDataMessage into chunks.

func (*Fragmenter) FragmentWriteRequest 

func (f *Fragmenter) FragmentWriteRequest(msg *message.WriteRequestMessage) ([]*message.WriteRequestMessage, error)

FragmentWriteRequest splits a WriteRequestMessage into chunks.

func (*Fragmenter) NeedsChunking 

func (f *Fragmenter) NeedsChunking(data []byte) bool

NeedsChunking returns true if the message exceeds the MTU.

type InvokeCall 

type InvokeCall struct {
	Path    imsg.CommandPathIB
	IsTimed bool
	Fields  []byte
}

InvokeCall records a command invocation.

type InvokeContext 

type InvokeContext struct {
	// Exchange is the underlying exchange context.
	Exchange *exchange.ExchangeContext

	// FabricIndex is the accessing fabric (0 if none).
	FabricIndex uint8

	// IsTimed indicates if this is part of a timed interaction.
	IsTimed bool

	// SourceNodeID is the requesting node.
	SourceNodeID uint64
}

InvokeContext provides context for command invocation.

type InvokeHandler 

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

InvokeHandler handles invoke request messages. It supports chunked requests and responses for large payloads.

func NewInvokeHandler 

func NewInvokeHandler(handler CommandHandler, maxPayload int, log logging.LeveledLogger) *InvokeHandler

NewInvokeHandler creates a new invoke handler.

func (*InvokeHandler) HandleInvokeRequest 

func (h *InvokeHandler) HandleInvokeRequest(
	exchCtx *exchange.ExchangeContext,
	msg *message.InvokeRequestMessage,
	fabricIndex uint8,
	sourceNodeID uint64,
	isTimed bool,
) (*message.InvokeResponseMessage, error)

HandleInvokeRequest processes an incoming InvokeRequestMessage. Returns the response message (or nil for chunked flow control).

func (*InvokeHandler) HandleStatusResponse 

func (h *InvokeHandler) HandleStatusResponse(status message.Status) (*message.InvokeResponseMessage, error)

HandleStatusResponse processes a StatusResponse during chunked transmission. Returns the next response chunk, or nil if transmission is complete.

func (*InvokeHandler) Reset 

func (h *InvokeHandler) Reset()

Reset resets the handler to idle state.

func (*InvokeHandler) State 

func (h *InvokeHandler) State() InvokeHandlerState

State returns the current handler state.

type InvokeHandlerState 

type InvokeHandlerState int

InvokeHandlerState represents the handler state machine. 

const (
	InvokeHandlerStateIdle InvokeHandlerState = iota
	InvokeHandlerStateReceiving
	InvokeHandlerStateProcessing
	InvokeHandlerStateSendingResponse
)

func (InvokeHandlerState) String 

func (s InvokeHandlerState) String() string

String returns the state name.

type InvokeResult 

type InvokeResult struct {
	// ResponseData is the TLV-encoded response fields.
	ResponseData []byte

	// Status is the IM status if command returned status instead of data.
	Status imsg.Status

	// ClusterStatus is the cluster-specific status code if present.
	ClusterStatus *uint16

	// HasStatus indicates whether this is a status response vs data response.
	HasStatus bool
}

InvokeResult is the result of an invoke operation.

type MockDispatcher 

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

MockDispatcher records IM operations for testing verification. Thread-safe for concurrent access.

func NewMockDispatcher 

func NewMockDispatcher() *MockDispatcher

NewMockDispatcher creates a new mock dispatcher.

func (*MockDispatcher) InvokeCalls 

func (d *MockDispatcher) InvokeCalls() []InvokeCall

InvokeCalls returns recorded invoke calls.

func (*MockDispatcher) InvokeCommand 

func (d *MockDispatcher) InvokeCommand(ctx context.Context, req *CommandInvokeRequest, r *tlv.Reader) ([]byte, error)

InvokeCommand implements Dispatcher.

func (*MockDispatcher) ReadAttribute 

func (d *MockDispatcher) ReadAttribute(ctx context.Context, req *AttributeReadRequest, w *tlv.Writer) error

ReadAttribute implements Dispatcher.

func (*MockDispatcher) ReadCalls 

func (d *MockDispatcher) ReadCalls() []ReadCall

ReadCalls returns recorded read calls.

func (*MockDispatcher) Reset 

func (d *MockDispatcher) Reset()

Reset clears all recorded calls.

func (*MockDispatcher) SetInvokeResult 

func (d *MockDispatcher) SetInvokeResult(response []byte, err error)

SetInvokeResult configures the response for InvokeCommand.

func (*MockDispatcher) SetReadResult 

func (d *MockDispatcher) SetReadResult(value interface{}, err error)

SetReadResult configures the response for ReadAttribute. Value will be encoded as TLV.

func (*MockDispatcher) SetWriteResult 

func (d *MockDispatcher) SetWriteResult(err error)

SetWriteResult configures the response for WriteAttribute.

func (*MockDispatcher) WriteAttribute 

func (d *MockDispatcher) WriteAttribute(ctx context.Context, req *AttributeWriteRequest, r *tlv.Reader) error

WriteAttribute implements Dispatcher.

func (*MockDispatcher) WriteCalls 

func (d *MockDispatcher) WriteCalls() []WriteCall

WriteCalls returns recorded write calls.

type MockInvokeResult 

type MockInvokeResult struct {
	Response []byte
	Err      error
}

MockInvokeResult configures the response for InvokeCommand.

type MockReadResult 

type MockReadResult struct {
	Value interface{} // Written to TLV
	Err   error
}

MockReadResult configures the response for ReadAttribute.

type NullDispatcher 

type NullDispatcher struct{}

NullDispatcher is a dispatcher that returns UnsupportedCluster for all operations. Use as a placeholder when no data model is available.

func (NullDispatcher) InvokeCommand 

func (NullDispatcher) InvokeCommand(ctx context.Context, req *CommandInvokeRequest, r *tlv.Reader) ([]byte, error)

InvokeCommand always returns ErrClusterNotFound.

func (NullDispatcher) ReadAttribute 

func (NullDispatcher) ReadAttribute(ctx context.Context, req *AttributeReadRequest, w *tlv.Writer) error

ReadAttribute always returns ErrClusterNotFound.

func (NullDispatcher) WriteAttribute 

func (NullDispatcher) WriteAttribute(ctx context.Context, req *AttributeWriteRequest, r *tlv.Reader) error

WriteAttribute always returns ErrClusterNotFound.

type ReadCall 

type ReadCall struct {
	Path             imsg.AttributePathIB
	IsFabricFiltered bool
}

ReadCall records an attribute read.

type ReadContext 

type ReadContext struct {
	// Exchange is the underlying exchange context.
	Exchange *exchange.ExchangeContext

	// FabricIndex is the accessing fabric (0 if none).
	FabricIndex uint8

	// IsFabricFiltered indicates fabric-filtered read.
	IsFabricFiltered bool

	// SourceNodeID is the requesting node.
	SourceNodeID uint64
}

ReadContext provides context for attribute reads.

type ReadHandler 

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

ReadHandler handles read request messages. This is a simplified implementation for Descriptor/Basic clusters. It does NOT support:

  • Wildcard path expansion (concrete paths only)
  • Complex ACL checks (assumes caller validated access)
  • Chunked report assembly (single response)

For full IM spec compliance, see docs/pkgs/im-plan.md.

func NewReadHandler 

func NewReadHandler(reader AttributeReader, maxPayload int) *ReadHandler

NewReadHandler creates a new read handler.

func (*ReadHandler) HandleReadRequest 

func (h *ReadHandler) HandleReadRequest(
	exchCtx *exchange.ExchangeContext,
	msg *message.ReadRequestMessage,
	fabricIndex uint8,
	sourceNodeID uint64,
) (*message.ReportDataMessage, error)

HandleReadRequest processes an incoming ReadRequestMessage. Returns the ReportData response message.

func (*ReadHandler) HandleStatusResponse 

func (h *ReadHandler) HandleStatusResponse(status message.Status) (*message.ReportDataMessage, error)

HandleStatusResponse processes a StatusResponse during chunked transmission.

func (*ReadHandler) Reset 

func (h *ReadHandler) Reset()

Reset resets the handler to idle state.

func (*ReadHandler) State 

func (h *ReadHandler) State() ReadHandlerState

State returns the current handler state.

type ReadHandlerState 

type ReadHandlerState int

ReadHandlerState represents the handler state machine. 

const (
	ReadHandlerStateIdle ReadHandlerState = iota
	ReadHandlerStateProcessing
	ReadHandlerStateSendingReport
)

func (ReadHandlerState) String 

func (s ReadHandlerState) String() string

String returns the state name.

type RequestContext 

type RequestContext struct {
	// Exchange is the underlying exchange context.
	// Provides access to session info and message sending.
	Exchange *exchange.ExchangeContext

	// Subject describes the identity making the request.
	// Used for ACL validation and fabric-scoped data access.
	Subject acl.SubjectDescriptor
}

RequestContext provides context for IM operations. It wraps the exchange context and provides subject descriptor for ACL checks. This is passed to all handler operations and can be used by clusters.

func NewRequestContext 

func NewRequestContext(exchCtx *exchange.ExchangeContext, subject acl.SubjectDescriptor) *RequestContext

NewRequestContext creates a new request context.

func (*RequestContext) AuthMode 

func (c *RequestContext) AuthMode() acl.AuthMode

AuthMode returns the authentication mode of the session.

func (*RequestContext) FabricIndex 

func (c *RequestContext) FabricIndex() fabric.FabricIndex

FabricIndex returns the accessing fabric index.

func (*RequestContext) IsCommissioning 

func (c *RequestContext) IsCommissioning() bool

IsCommissioning returns true if this is during PASE commissioning. During commissioning, implicit Administer privilege is granted.

func (*RequestContext) SourceNodeID 

func (c *RequestContext) SourceNodeID() uint64

SourceNodeID returns the requesting node's ID.

type SecureTestIMPair 

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

SecureTestIMPair provides two connected IM engines with encrypted sessions.

func NewSecureTestIMPair 

func NewSecureTestIMPair(config SecureTestIMPairConfig) (*SecureTestIMPair, error)

NewSecureTestIMPair creates a new paired IM test environment with secure sessions.

func (*SecureTestIMPair) Client 

func (p *SecureTestIMPair) Client(idx int) *Client

Client returns the IM client at the given index.

func (*SecureTestIMPair) Close 

func (p *SecureTestIMPair) Close()

Close releases resources.

func (*SecureTestIMPair) Engine 

func (p *SecureTestIMPair) Engine(idx int) *Engine

Engine returns the IM engine at the given index.

func (*SecureTestIMPair) ExchangePair 

func (p *SecureTestIMPair) ExchangePair() *exchange.TestManagerPair

ExchangePair returns the underlying exchange test pair.

func (*SecureTestIMPair) PeerAddress 

func (p *SecureTestIMPair) PeerAddress(idx int) transport.PeerAddress

PeerAddress returns the peer address for the given index.

func (*SecureTestIMPair) Session 

func (p *SecureTestIMPair) Session(idx int) *session.SecureContext

Session returns the secure session for the given index.

type SecureTestIMPairConfig 

type SecureTestIMPairConfig struct {
	// Dispatcher for each side (index 0 = client side, index 1 = server side)
	Dispatchers [2]Dispatcher
}

SecureTestIMPairConfig configures a SecureTestIMPair.

type TLVMarshaler 

type TLVMarshaler interface {
	MarshalTLV(w *tlv.Writer) error
}

TLVMarshaler is implemented by types that can marshal themselves to TLV. Event payload structs should implement this interface.

type TestIMPair 

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

TestIMPair provides two connected IM engines for E2E testing. The client side (index 0) can send requests to the server side (index 1).

Architecture: 

Client (0)                          Server (1)
──────────                          ──────────
im.Client                           im.Engine
    │                                   │
    ▼                                   ▼
exchange.Manager ◀──── Pipe ────▶ exchange.Manager

func NewTestIMPair 

func NewTestIMPair(config TestIMPairConfig) (*TestIMPair, error)

NewTestIMPair creates a new paired IM test environment.

func (*TestIMPair) Client 

func (p *TestIMPair) Client(idx int) *Client

Client returns the IM client at the given index.

func (*TestIMPair) Close 

func (p *TestIMPair) Close()

Close releases resources.

func (*TestIMPair) Engine 

func (p *TestIMPair) Engine(idx int) *Engine

Engine returns the IM engine at the given index.

func (*TestIMPair) ExchangePair 

func (p *TestIMPair) ExchangePair() *exchange.TestManagerPair

ExchangePair returns the underlying exchange test pair.

func (*TestIMPair) PeerAddress 

func (p *TestIMPair) PeerAddress(idx int) transport.PeerAddress

PeerAddress returns the peer address for the given index.

func (*TestIMPair) Session 

func (p *TestIMPair) Session(idx int) *exchange.TestUnsecuredSession

Session returns a test session for the given index.

type TestIMPairConfig 

type TestIMPairConfig struct {
	// Dispatcher for each side (index 0 = client side, index 1 = server side)
	// If nil, NullDispatcher is used.
	Dispatchers [2]Dispatcher
}

TestIMPairConfig configures a TestIMPair.

type WriteCall 

type WriteCall struct {
	Path    imsg.AttributePathIB
	IsTimed bool
	Data    []byte
}

WriteCall records an attribute write.

type WriteContext 

type WriteContext struct {
	// Exchange is the underlying exchange context.
	Exchange *exchange.ExchangeContext

	// FabricIndex is the accessing fabric (0 if none).
	FabricIndex uint8

	// IsTimed indicates if this is part of a timed interaction.
	IsTimed bool

	// SourceNodeID is the requesting node.
	SourceNodeID uint64
}

WriteContext provides context for attribute writes.

type WriteHandler 

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

WriteHandler handles write request messages. This is a simplified implementation for commissioning use cases. It does NOT support:

  • Wildcard paths (concrete paths only)
  • Chunked write requests (single message only)
  • List operations (Add/Remove - only full Replace)

For full IM spec compliance, see docs/pkgs/im-plan.md.

Spec Reference: Section 8.7 "Write Interaction" C++ Reference: src/app/WriteHandler.cpp

func NewWriteHandler 

func NewWriteHandler(dispatcher Dispatcher) *WriteHandler

NewWriteHandler creates a new write handler.

func (*WriteHandler) HandleWriteRequest 

func (h *WriteHandler) HandleWriteRequest(
	exchCtx *exchange.ExchangeContext,
	msg *message.WriteRequestMessage,
	fabricIndex uint8,
	sourceNodeID uint64,
	isTimed bool,
) (*message.WriteResponseMessage, error)

HandleWriteRequest processes an incoming WriteRequestMessage. Returns the WriteResponseMessage.

Spec: 8.7.3.2 "Outgoing Write Response Action" (server-side processing)

func (*WriteHandler) Reset 

func (h *WriteHandler) Reset()

Reset resets the handler to idle state.

func (*WriteHandler) State 

func (h *WriteHandler) State() WriteHandlerState

State returns the current handler state.

type WriteHandlerState 

type WriteHandlerState int

WriteHandlerState represents the handler state machine. Spec: 8.7 Write Interaction 

const (
	WriteHandlerStateIdle WriteHandlerState = iota
	WriteHandlerStateProcessing
	WriteHandlerStateReceivingChunks
	WriteHandlerStateSendingResponse
)

func (WriteHandlerState) String 

func (s WriteHandlerState) String() string

String returns the state name.

Source Files

View all Source files

Directories

Path Synopsis

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.
⌂ 换网址