Documentation
¶
Index ¶
- Variables
- func IsCancellationUnacknowledged(err error) bool
- type BidiStream
- type CancellationMessageEnvelope
- type MessageBroker
- func (mb *MessageBroker[TMessage]) Close()
- func (mb *MessageBroker[TMessage]) Done() <-chan struct{}
- func (mb *MessageBroker[TMessage]) On(handler any) error
- func (mb *MessageBroker[TMessage]) Ready(ctx context.Context) error
- func (mb *MessageBroker[TMessage]) Run(ctx context.Context) error
- func (mb *MessageBroker[TMessage]) Send(ctx context.Context, msg *TMessage) error
- func (mb *MessageBroker[TMessage]) SendAndWait(ctx context.Context, msg *TMessage) (*TMessage, error)
- func (mb *MessageBroker[TMessage]) SendAndWaitWithProgress(ctx context.Context, msg *TMessage, onProgress func(string)) (*TMessage, error)
- type MessageBrokerOption
- type MessageEnvelope
- type PersistentHandlerContextEnvelope
- type ProgressFunc
Constants ¶
This section is empty.
Variables ¶
var ErrResourceExhausted = errors.New("gRPC message size limit exceeded")
ErrResourceExhausted is returned when a gRPC message exceeds the size limit.
Functions ¶
func IsCancellationUnacknowledged ¶ added in v1.37.0
IsCancellationUnacknowledged reports whether the broker returned without confirming that the canceled remote request had completed.
Types ¶
type BidiStream ¶
type BidiStream[TMessage any] interface { // Send sends a message on the stream Send(msg *TMessage) error // Recv receives a message from the stream Recv() (*TMessage, error) }
BidiStream is a unified interface for bidirectional gRPC streams. Both grpc.BidiStreamingClient[T, T] and grpc.BidiStreamingServer[T, T] implement this interface, allowing the MessageBroker to work with either client-side or server-side streams.
type CancellationMessageEnvelope ¶ added in v1.37.0
type CancellationMessageEnvelope[T any] interface { CreateCancellationMessage(ctx context.Context, request *T, err error) *T IsCancellationMessage(ctx context.Context, msg *T) bool GetCancellationError(msg *T) error }
CancellationMessageEnvelope customizes cancellation control messages for envelopes that cannot use the default request-id + error representation.
type MessageBroker ¶
type MessageBroker[TMessage any] struct { // contains filtered or unexported fields }
MessageBroker handles bidirectional message routing for gRPC streams. It supports both client pattern (request/response correlation via RequestId) and server pattern (handler registration for incoming requests).
TMessage is the raw message type used by the gRPC stream. The ops parameter provides stateless operations for manipulating messages.
This broker works with both client-side (grpc.BidiStreamingClient) and server-side (grpc.BidiStreamingServer) streams through the unified BidiStream interface.
func NewMessageBroker ¶
func NewMessageBroker[TMessage any]( stream BidiStream[TMessage], ops MessageEnvelope[TMessage], name string, logger *log.Logger, ) *MessageBroker[TMessage]
NewMessageBroker creates a new message broker for the given stream. The stream parameter can be either a client stream (grpc.BidiStreamingClient) or a server stream (grpc.BidiStreamingServer) as both implement the BidiStream interface. The ops parameter provides stateless operations for message manipulation. The name parameter is used for logging identification. The logger parameter sets the broker's private logger for trace output: Pass log.Default on the server side (azd core CLI) to inherit --debug semantics. Pass nil for silent operation (e.g., in extension processes where AZD_EXT_DEBUG controls logging).
func NewMessageBrokerWithOptions ¶ added in v1.37.0
func NewMessageBrokerWithOptions[TMessage any]( stream BidiStream[TMessage], ops MessageEnvelope[TMessage], name string, logger *log.Logger, options ...MessageBrokerOption, ) *MessageBroker[TMessage]
NewMessageBrokerWithOptions creates a new message broker and applies the provided options.
func (*MessageBroker[TMessage]) Close ¶
func (mb *MessageBroker[TMessage]) Close()
Close gracefully shuts down the broker (optional, for cleanup)
func (*MessageBroker[TMessage]) Done ¶ added in v1.37.0
func (mb *MessageBroker[TMessage]) Done() <-chan struct{}
Done returns a channel that is closed after the broker shuts down.
func (*MessageBroker[TMessage]) On ¶
func (mb *MessageBroker[TMessage]) On(handler any) error
On registers a handler for a specific message type. The handler function signature should be one of:
- func(ctx context.Context, req *RequestType) (*TMessage, error)
- func(ctx context.Context, req *RequestType, progress ProgressFunc) (*TMessage, error)
The handler must return a complete envelope message. The broker will automatically set the RequestId and Error fields before sending the response.
func (*MessageBroker[TMessage]) Ready ¶
func (mb *MessageBroker[TMessage]) Ready(ctx context.Context) error
Ready blocks until the message broker starts receiving messages or the context is cancelled. Multiple goroutines can call Ready() simultaneously - they will all be unblocked when Run() starts. Once the broker has started, subsequent calls to Ready() return immediately. Returns nil when ready, or context error if the context is cancelled before the broker becomes ready.
func (*MessageBroker[TMessage]) Run ¶
func (mb *MessageBroker[TMessage]) Run(ctx context.Context) error
Run begins receiving and dispatching messages. This method blocks until the context is cancelled, the stream encounters an error, or the stream is closed by the remote peer. Returns nil on graceful shutdown (context cancelled or EOF), or the error that terminated the stream.
func (*MessageBroker[TMessage]) Send ¶
func (mb *MessageBroker[TMessage]) Send(ctx context.Context, msg *TMessage) error
Send sends a message without waiting for a response. This is useful for fire-and-forget scenarios like subscriptions or notifications where no response is expected or needed. Returns an error only if the send operation itself fails.
func (*MessageBroker[TMessage]) SendAndWait ¶
func (mb *MessageBroker[TMessage]) SendAndWait(ctx context.Context, msg *TMessage) (*TMessage, error)
SendAndWait sends a message and waits for the response
func (*MessageBroker[TMessage]) SendAndWaitWithProgress ¶
func (mb *MessageBroker[TMessage]) SendAndWaitWithProgress( ctx context.Context, msg *TMessage, onProgress func(string), ) (*TMessage, error)
SendAndWaitWithProgress sends a message and waits for the response, handling progress updates
type MessageBrokerOption ¶ added in v1.37.0
type MessageBrokerOption func(*messageBrokerOptions)
MessageBrokerOption configures a MessageBroker.
func WithCancellationGracePeriod ¶ added in v1.37.0
func WithCancellationGracePeriod(gracePeriod time.Duration) MessageBrokerOption
WithCancellationGracePeriod controls how long a canceled caller waits for the remote handler to acknowledge cancellation before returning.
type MessageEnvelope ¶
type MessageEnvelope[T any] interface { // GetRequestId extracts or generates the request/correlation ID from a message. // For messages with a RequestId field, this returns that field. // For messages without RequestId, this can generate a correlation key from message content and context. GetRequestId(ctx context.Context, msg *T) string // SetRequestId sets the request ID on a message. // For messages without a RequestId field, this can be a no-op. SetRequestId(ctx context.Context, msg *T, id string) // GetError extracts the error from a message, if any GetError(msg *T) error // SetError sets an error on a message SetError(msg *T, err error) // GetInnerMessage extracts the inner message from the envelope's oneof field GetInnerMessage(msg *T) any // IsProgressMessage returns true if the message is a progress message IsProgressMessage(msg *T) bool // GetProgressMessage extracts the progress message text from a progress message. // Returns empty string if the message is not a progress message. GetProgressMessage(msg *T) string // CreateProgressMessage creates a new progress message envelope with the given text. // This is used by server-side handlers to send progress updates back to clients. CreateProgressMessage(requestId string, message string) *T }
MessageEnvelope provides broker-specific operations on message types. This is a stateless service that knows how to extract and manipulate message fields. The methods work with pointers (*T) to avoid copying and to match gRPC's pointer-based APIs. Context is provided to allow envelopes to extract metadata (e.g., extension claims from gRPC context).
type PersistentHandlerContextEnvelope ¶ added in v1.37.0
type PersistentHandlerContextEnvelope[T any] interface { PreserveHandlerContext(ctx context.Context, msg *T) bool }
PersistentHandlerContextEnvelope identifies messages whose handlers retain the stream context after returning, such as event subscriptions. These handlers must not receive a request-scoped context that is canceled as soon as the subscription call completes.
type ProgressFunc ¶
type ProgressFunc func(message string)
ProgressFunc is a callback function for sending progress updates during handler execution
Source Files