amqp

package
v1.6.0 Latest Latest
Warning

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

 Go to latest Published: Sep 2, 2025 License: MIT Imports: 10 Imported by: 25

Documentation

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

This section is empty.

Types

type AMQP10 

type AMQP10 struct {
	Properties            *MessageProperties
	Annotations           Annotations
	ApplicationProperties map[string]any
	// contains filtered or unexported fields
}

AMQP10 is an AMQP 1.0 message with the necessary fields to work with the stream package. This is a wrapper around the amqp.Message struct. AMQP10 is not thread-safe. You should avoid to share the same message between multiple go routines and between multi Send/BatchSend calls. Each Send/BatchSend call should use a new message.

func NewMessage 

func NewMessage(data []byte) *AMQP10

func (*AMQP10) GetAMQPValue 

func (amqp *AMQP10) GetAMQPValue() any

func (*AMQP10) GetApplicationProperties 

func (amqp *AMQP10) GetApplicationProperties() map[string]any

func (*AMQP10) GetData 

func (amqp *AMQP10) GetData() [][]byte

func (*AMQP10) GetMessageAnnotations 

func (amqp *AMQP10) GetMessageAnnotations() Annotations

func (*AMQP10) GetMessageHeader 

func (amqp *AMQP10) GetMessageHeader() *MessageHeader

func (*AMQP10) GetMessageProperties 

func (amqp *AMQP10) GetMessageProperties() *MessageProperties

func (*AMQP10) GetPublishingId 

func (amqp *AMQP10) GetPublishingId() int64

func (*AMQP10) HasPublishingId 

func (amqp *AMQP10) HasPublishingId() bool

func (*AMQP10) MarshalBinary 

func (amqp *AMQP10) MarshalBinary() ([]byte, error)

func (*AMQP10) SetPublishingId 

func (amqp *AMQP10) SetPublishingId(id int64)

func (*AMQP10) UnmarshalBinary 

func (amqp *AMQP10) UnmarshalBinary(data []byte) error

type Annotations 

type Annotations map[any]any

Annotations keys must be of type string, int, or int64.

String keys are encoded as AMQP Symbols.

type ArrayUByte 

type ArrayUByte []uint8

ArrayUByte allows encoding []uint8/[]byte as an array rather than binary data.

type Durability 

type Durability uint32

Durability specifies the durability of a link. 

const (
	// No terminus state is retained durably.
	DurabilityNone Durability = 0

	// Only the existence and configuration of the terminus is
	// retained durably.
	DurabilityConfiguration Durability = 1

	// In addition to the existence and configuration of the
	// terminus, the unsettled state for durable messages is
	// retained durably.
	DurabilityUnsettledState Durability = 2
)

Durability Policies

func (*Durability) String 

func (d *Durability) String() string

type Error 

type Error struct {
	// A symbolic value indicating the error condition.
	Condition ErrorCondition

	// descriptive text about the error condition
	//
	// This text supplies any supplementary details not indicated by the condition field.
	// This text can be logged as an aid to resolving issues.
	Description string

	// map carrying information about the error condition
	Info map[string]any
}

Error is an AMQP error.

func (*Error) Error 

func (e *Error) Error() string

func (*Error) String 

func (e *Error) String() string

type ErrorCondition 

type ErrorCondition string

ErrorCondition is one of the error conditions defined in the AMQP spec. 

const (
	// AMQP Errors
	ErrorInternalError         ErrorCondition = "amqp:internal-error"
	ErrorNotFound              ErrorCondition = "amqp:not-found"
	ErrorUnauthorizedAccess    ErrorCondition = "amqp:unauthorized-access"
	ErrorDecodeError           ErrorCondition = "amqp:decode-error"
	ErrorResourceLimitExceeded ErrorCondition = "amqp:resource-limit-exceeded"
	ErrorNotAllowed            ErrorCondition = "amqp:not-allowed"
	ErrorInvalidField          ErrorCondition = "amqp:invalid-field"
	ErrorNotImplemented        ErrorCondition = "amqp:not-implemented"
	ErrorResourceLocked        ErrorCondition = "amqp:resource-locked"
	ErrorPreconditionFailed    ErrorCondition = "amqp:precondition-failed"
	ErrorResourceDeleted       ErrorCondition = "amqp:resource-deleted"
	ErrorIllegalState          ErrorCondition = "amqp:illegal-state"
	ErrorFrameSizeTooSmall     ErrorCondition = "amqp:frame-size-too-small"

	// Connection Errors
	ErrorConnectionForced   ErrorCondition = "amqp:connection:forced"
	ErrorFramingError       ErrorCondition = "amqp:connection:framing-error"
	ErrorConnectionRedirect ErrorCondition = "amqp:connection:redirect"

	// Session Errors
	ErrorWindowViolation  ErrorCondition = "amqp:session:window-violation"
	ErrorErrantLink       ErrorCondition = "amqp:session:errant-link"
	ErrorHandleInUse      ErrorCondition = "amqp:session:handle-in-use"
	ErrorUnattachedHandle ErrorCondition = "amqp:session:unattached-handle"

	// Link Errors
	ErrorDetachForced          ErrorCondition = "amqp:link:detach-forced"
	ErrorTransferLimitExceeded ErrorCondition = "amqp:link:transfer-limit-exceeded"
	ErrorMessageSizeExceeded   ErrorCondition = "amqp:link:message-size-exceeded"
	ErrorLinkRedirect          ErrorCondition = "amqp:link:redirect"
	ErrorStolen                ErrorCondition = "amqp:link:stolen"
)

Error Conditions

type ExpiryPolicy 

type ExpiryPolicy symbol

ExpiryPolicy specifies when the expiry timer of a terminus starts counting down from the timeout value.

If the link is subsequently re-attached before the terminus is expired, then the count down is aborted. If the conditions for the terminus-expiry-policy are subsequently re-met, the expiry timer restarts from its originally configured timeout value. 

const (
	// The expiry timer starts when terminus is detached.
	ExpiryLinkDetach ExpiryPolicy = "link-detach"

	// The expiry timer starts when the most recently
	// associated session is ended.
	ExpirySessionEnd ExpiryPolicy = "session-end"

	// The expiry timer starts when most recently associated
	// connection is closed.
	ExpiryConnectionClose ExpiryPolicy = "connection-close"

	// The terminus never expires.
	ExpiryNever ExpiryPolicy = "never"
)

Expiry Policies

func (*ExpiryPolicy) String 

func (e *ExpiryPolicy) String() string

type Message 

type Message struct {
	// Message format code.
	//
	// The upper three octets of a message format code identify a particular message
	// format. The lowest octet indicates the version of said message format. Any
	// given version of a format is forwards compatible with all higher versions.
	Format uint32

	// The DeliveryTag can be up to 32 octets of binary data.
	// Note that when mode one is enabled there will be no delivery tag.
	DeliveryTag []byte

	// The header section carries standard delivery details about the transfer
	// of a message through the AMQP network.
	Header *MessageHeader

	// The delivery-annotations section is used for delivery-specific non-standard
	// properties at the head of the message. Delivery annotations convey information
	// from the sending peer to the receiving peer.
	DeliveryAnnotations Annotations

	// The message-annotations section is used for properties of the message which
	// are aimed at the infrastructure.
	Annotations Annotations

	// The properties section is used for a defined set of standard properties of
	// the message.
	Properties *MessageProperties

	// The application-properties section is a part of the bare message used for
	// structured application data. Intermediaries can use the data within this
	// structure for the purposes of filtering or routing.
	ApplicationProperties map[string]any

	// Data payloads.
	Data [][]byte

	// Value payload.
	Value any

	// The footer section is used for details about the message or delivery which
	// can only be calculated or evaluated once the whole bare message has been
	// constructed or seen (for example message hashes, HMACs, signatures and
	// encryption details).
	Footer Annotations

	// Mark the message as settled when LinkSenderSettle is ModeMixed.
	//
	// This field is ignored when LinkSenderSettle is not ModeMixed.
	SendSettled bool
}

Message is an AMQP message.

func (*Message) GetData 

func (m *Message) GetData() []byte

GetData returns the first []byte from the Data field or nil if Data is empty.

func (*Message) MarshalBinary 

func (m *Message) MarshalBinary() ([]byte, error)

MarshalBinary encodes the message into binary form.

func (*Message) UnmarshalBinary 

func (m *Message) UnmarshalBinary(data []byte) error

UnmarshalBinary decodes the message from binary form.

type MessageHeader 

type MessageHeader struct {
	Durable       bool
	Priority      uint8
	TTL           time.Duration // from milliseconds
	FirstAcquirer bool
	DeliveryCount uint32
}

MessageHeader carries standard delivery details about the transfer of a message.

type MessageProperties 

type MessageProperties struct {
	// Message-id, if set, uniquely identifies a message within the message system.
	// The message producer is usually responsible for setting the message-id in
	// such a way that it is assured to be globally unique. A broker MAY discard a
	// message as a duplicate if the value of the message-id matches that of a
	// previously received message sent to the same node.
	MessageID any // uint64, UUID, []byte, or string

	// The identity of the user responsible for producing the message.
	// The client sets this value, and it MAY be authenticated by intermediaries.
	UserID []byte

	// The to field identifies the node that is the intended destination of the message.
	// On any given transfer this might not be the node at the receiving end of the link.
	To string

	// A common field for summary information about the message content and purpose.
	Subject string

	// The address of the node to send replies to.
	ReplyTo string

	// This is a client-specific id that can be used to mark or identify messages
	// between clients.
	CorrelationID any // uint64, UUID, []byte, or string

	// The RFC-2046 [RFC2046] MIME type for the message's application-data section
	// (body). As per RFC-2046 [RFC2046] this can contain a charset parameter defining
	// the character encoding used: e.g., 'text/plain; charset="utf-8"'.
	//
	// For clarity, as per section 7.2.1 of RFC-2616 [RFC2616], where the content type
	// is unknown the content-type SHOULD NOT be set. This allows the recipient the
	// opportunity to determine the actual type. Where the section is known to be truly
	// opaque binary data, the content-type SHOULD be set to application/octet-stream.
	//
	// When using an application-data section with a section code other than data,
	// content-type SHOULD NOT be set.
	ContentType string

	// The content-encoding property is used as a modifier to the content-type.
	// When present, its value indicates what additional content encodings have been
	// applied to the application-data, and thus what decoding mechanisms need to be
	// applied in order to obtain the media-type referenced by the content-type header
	// field.
	//
	// Content-encoding is primarily used to allow a document to be compressed without
	// losing the identity of its underlying content type.
	//
	// Content-encodings are to be interpreted as per section 3.5 of RFC 2616 [RFC2616].
	// Valid content-encodings are registered at IANA [IANAHTTPPARAMS].
	//
	// The content-encoding MUST NOT be set when the application-data section is other
	// than data. The binary representation of all other application-data section types
	// is defined completely in terms of the AMQP type system.
	//
	// Implementations MUST NOT use the identity encoding. Instead, implementations
	// SHOULD NOT set this property. Implementations SHOULD NOT use the compress encoding,
	// except as to remain compatible with messages originally sent with other protocols,
	// e.g. HTTP or SMTP.
	//
	// Implementations SHOULD NOT specify multiple content-encoding values except as to
	// be compatible with messages originally sent with other protocols, e.g. HTTP or SMTP.
	ContentEncoding string

	// An absolute time when this message is considered to be expired.
	AbsoluteExpiryTime time.Time

	// An absolute time when this message was created.
	CreationTime time.Time

	// Identifies the group the message belongs to.
	GroupID string

	// The relative position of this message within its group.
	GroupSequence uint32 // RFC-1982 sequence number

	// This is a client-specific id that is used so that client can send replies to this
	// message to a specific group.
	ReplyToGroupID string
}

MessageProperties is the defined set of properties for AMQP messages.

type ProtoID 

type ProtoID uint8

type ReceiverSettleMode 

type ReceiverSettleMode uint8

ReceiverSettleMode specifies how the receiver will settle messages. 

const (
	// Receiver will spontaneously settle all incoming transfers.
	ModeFirst ReceiverSettleMode = 0

	// Receiver will only settle after sending the disposition to the
	// sender and receiving a disposition indicating settlement of
	// the delivery from the sender.
	ModeSecond ReceiverSettleMode = 1
)

Receiver Settlement Modes

func (*ReceiverSettleMode) String 

func (m *ReceiverSettleMode) String() string

type SenderSettleMode 

type SenderSettleMode uint8

SenderSettleMode specifies how the sender will settle messages. 

const (
	// Sender will send all deliveries initially unsettled to the receiver.
	ModeUnsettled SenderSettleMode = 0

	// Sender will send all deliveries settled to the receiver.
	ModeSettled SenderSettleMode = 1

	// Sender MAY send a mixture of settled and unsettled deliveries to the receiver.
	ModeMixed SenderSettleMode = 2
)

Sender Settlement Modes

func (*SenderSettleMode) String 

func (m *SenderSettleMode) String() string

type UUID 

type UUID [16]byte

UUID is a 128 bit identifier as defined in RFC 4122.

func (UUID) String 

func (u UUID) String() string

String returns the hex encoded representation described in RFC 4122, Section 3.

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.