AuthenticationError indicates an authentication related error occurred when connecting to a remote event broker. The pointer type *AuthenticationError is returned.
type AuthenticationError struct {
// contains filtered or unexported fields
}
func (err *AuthenticationError) Error() string
Error returns the error message.
func (err *AuthenticationError) Unwrap() error
Unwrap returns the wrapped error.
The DirectMessagePublisher interface is used to publish direct messages.
type DirectMessagePublisher interface {
MessagePublisher
MessagePublisherHealthCheck
// StartAsyncCallback starts the DirectMessagePublisher asynchronously.
// Calls the specified callback when started with an error if one occurred, otherwise nil
// if successful.
StartAsyncCallback(callback func(DirectMessagePublisher, error))
// SetPublishFailureListener sets the listener to call if the publishing of
// a direct message fails.
SetPublishFailureListener(listener PublishFailureListener)
// TerminateAsyncCallback terminates the DirectMessagePublisher asynchronously.
// Calls the callback when terminated with nil if successful, or an error if
// one occurred. If gracePeriod is less than 0, this function waits indefinitely.
TerminateAsyncCallback(gracePeriod time.Duration, callback func(error))
// PublishBytes publishes a message of type byte array to the specified destination.
// Returns an error if one occurred while attempting to publish, or if the publisher
// is not started/terminated. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than the publisher's I/O
// capabilities allow. When publishing can be resumed, the registered
// PublisherReadinessListeners are called.
PublishBytes(message []byte, destination *resource.Topic) error
// PublishString publishes a message of type string to the specified destination.
// Returns an error if one occurred. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than the publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners are called.
PublishString(message string, destination *resource.Topic) error
// Publish publishes the specified message of type OutboundMessage built by a
// OutboundMessageBuilder to the specified destination. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than the publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners are called.
Publish(message message.OutboundMessage, destination *resource.Topic) error
// PublishWithProperties publishes the specified message of type OutboundMessage
// with the specified properties. These properties override the properties on
// the OutboundMessage instance if it is present. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than the publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners are called.
PublishWithProperties(message message.OutboundMessage, destination *resource.Topic, properties config.MessagePropertiesConfigurationProvider) error
}
DirectMessagePublisherBuilder allows for configuration of direct message publisher instances.
type DirectMessagePublisherBuilder interface {
// Build creates a new DirectMessagePublisher instance based on the configured properties.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build() (messagePublisher DirectMessagePublisher, err error)
// OnBackPressureReject sets the publisher back pressure strategy to reject
// where publish attempts will be rejected once the bufferSize, in number of messages, is reached.
// If bufferSize is 0, an error will be thrown when the transport is full when publishing.
// A buffer of the given size will be statically allocated when the publisher is built.
// Valid bufferSize is >= 0.
OnBackPressureReject(bufferSize uint) DirectMessagePublisherBuilder
// OnBackPressureWait sets the publisher back pressure strategy to wait where publish
// attempts may block until there is space in the buffer of size bufferSize in number of messages.
// A buffer of the given size will be statically allocated when the publisher is built.
// Valid bufferSize is >= 1.
OnBackPressureWait(bufferSize uint) DirectMessagePublisherBuilder
// FromConfigurationProvider configures the direct publisher with the specified properties.
// The built-in PublisherPropertiesConfigurationProvider implementations include:
// - PublisherPropertyMap - A map of PublisherProperty keys to values.
FromConfigurationProvider(provider config.PublisherPropertiesConfigurationProvider) DirectMessagePublisherBuilder
}
The DirectMessageReceiver is used to receive direct messages.
type DirectMessageReceiver interface {
MessageReceiver // Include all functionality of MessageReceiver.
// StartAsyncCallback starts the DirectMessageReceiver asynchronously.
// Calls the callback when started with an error if one occurred, otherwise nil
// if successful.
StartAsyncCallback(callback func(DirectMessageReceiver, error))
// TerminateAsyncCallback terminates the DirectMessageReceiver asynchronously.
// Calls the callback when terminated with nil if successful or an error if
// one occurred. If gracePeriod is less than 0, the function will wait indefinitely.
TerminateAsyncCallback(gracePeriod time.Duration, callback func(error))
// ReceiveAsync registers a callback to be called when new messages
// are received. Returns an error if one occurred while registering the callback.
// If a callback is already registered, it will be replaced by the specified
// callback.
ReceiveAsync(callback MessageHandler) error
// ReceiveMessage receives a inbound message synchronously from the receiver.
// Returns an error if the receiver has not started, or has already terminated.
// ReceiveMessage waits until the specified timeout to receive a message, or will wait
// forever if the timeout specified is a negative value. If a timeout occurs, a solace.TimeoutError
// is returned.
ReceiveMessage(timeout time.Duration) (received message.InboundMessage, err error)
}
DirectMessageReceiverBuilder allows for configuration of DirectMessageReceiver instances.
type DirectMessageReceiverBuilder interface {
// Build creates a DirectMessageReceiver with the specified properties.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build() (messageReceiver DirectMessageReceiver, err error)
// BuildWithShareName creates DirectMessageReceiver with the specified ShareName.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided
// or the specified ShareName is invalid.
BuildWithShareName(shareName *resource.ShareName) (messageReceiver DirectMessageReceiver, err error)
// OnBackPressureDropLatest configures the receiver with the specified buffer size. If the buffer
// is full and a message arrives, the incoming message is discarded.
// A buffer of the given size will be statically allocated when the receiver is built.
// The bufferCapacity must be greater than or equal to 1.
OnBackPressureDropLatest(bufferCapacity uint) DirectMessageReceiverBuilder
// OnBackPressureDropOldest configures the receiver with the specified buffer size, bufferCapacity. If the buffer
// is full and a message arrives, the oldest message in the buffer is discarded.
// A buffer of the given size will be statically allocated when the receiver is built.
// The value of bufferCapacity must be greater than or equal to 1.
OnBackPressureDropOldest(bufferCapacity uint) DirectMessageReceiverBuilder
// WithSubscriptions sets a list of TopicSubscriptions to subscribe
// to when starting the receiver. This function also accepts *resource.TopicSubscription subscriptions.
WithSubscriptions(topics ...resource.Subscription) DirectMessageReceiverBuilder
// FromConfigurationProvider configures the DirectMessageReceiver with the specified properties.
// The built-in ReceiverPropertiesConfigurationProvider implementations include:
// - ReceiverPropertyMap - A map of ReceiverProperty keys to values.
FromConfigurationProvider(provider config.ReceiverPropertiesConfigurationProvider) DirectMessageReceiverBuilder
}
Error is an error returned from the API.
type Error interface {
error
// contains filtered or unexported methods
}
func NewError(err Error, message string, wrapped error) Error
NewError returns a new Solace error with the specified message and wrapped error.
FailedPublishEvent represents an event thrown when publishing a direct message fails.
type FailedPublishEvent interface {
// GetMessage retrieves the message that was not delivered
GetMessage() message.OutboundMessage
// GetDestination retrieves the destination that the message was published to.
GetDestination() resource.Destination
// GetTimeStamp retrieves the timestamp of the error.
GetTimeStamp() time.Time
// GetError retrieves the error that failed the publish attempt.
GetError() error
}
IllegalArgumentError indicates an invalid argument was passed to a function. The pointer type *IllegalArgumentError is returned.
type IllegalArgumentError struct {
// contains filtered or unexported fields
}
func (err *IllegalArgumentError) Error() string
Error returns the error message.
func (err *IllegalArgumentError) Unwrap() error
Unwrap returns the wrapped error.
IllegalStateError indicates an invalid state occurred when performing an action. The pointer type *IllegalStateError is returned.
type IllegalStateError struct {
// contains filtered or unexported fields
}
func (err *IllegalStateError) Error() string
Error returns the error message.
func (err *IllegalStateError) Unwrap() error
Unwrap returns the wrapped error.
IncompleteMessageDeliveryError indicates that some messages were not delivered. The pointer type *IncompleteMessageDeliveryError is returned.
type IncompleteMessageDeliveryError struct {
// contains filtered or unexported fields
}
func (err *IncompleteMessageDeliveryError) Error() string
Error returns the error message.
func (err *IncompleteMessageDeliveryError) Unwrap() error
Unwrap returns the wrapped error.
InvalidConfigurationError indicates that a specified configuration is invalid. These errors are returned by the Build functions of a builder. The pointer type *InvalidConfigurationError is returned.
type InvalidConfigurationError struct {
// contains filtered or unexported fields
}
func (err *InvalidConfigurationError) Error() string
Error returns the error message.
func (err *InvalidConfigurationError) Unwrap() error
Unwrap returns the wrapped error.
LifecycleControl contains lifecycle functionality common to various messaging services such as publishers and receivers.
type LifecycleControl interface {
// Start starts the messaging service synchronously.
// Before this function is called, the messaging service is considered
// off-duty. To operate normally, this function must be called on
// a receiver or publisher instance. This function is idempotent.
// Returns an error if one occurred or nil if successful.
Start() error
// StartAsync starts the messaging service asynchronously.
// Before this function is called, the messaging service is considered
// off-duty. To operate normally, this function must be called on
// a receiver or publisher instance. This function is idempotent.
// Returns a channel that will receive an error if one occurred or
// nil if successful. Subsequent calls will return additional
// channels that can await an error, or nil if already started.
StartAsync() <-chan error
// Terminate terminates the messaging service gracefully and synchronously.
// This function is idempotent. The only way to resume operation
// after this function is called is to create another instance.
// Any attempt to call this function renders the instance
// permanently terminated, even if this function completes.
// A graceful shutdown is attempted within the specified grace period (gracePeriod).
// Setting gracePeriod to 0 implies a non-graceful shutdown that ignores
// unfinished tasks or in-flight messages.
// This function returns an error if one occurred, or
// nil if it successfully and gracefully terminated.
// If gracePeriod is set to less than 0, the function waits indefinitely.
Terminate(gracePeriod time.Duration) error
// TerminateAsync terminates the messaging service asynchronously.
// This function is idempotent. The only way to resume operation
// after this function is called is to create another instance.
// Any attempt to call this function renders the instance
// permanently terminated, even if this function completes.
// A graceful shutdown is attempted within the specified grace period (gracePeriod).
// Setting gracePeriod to 0 implies a non-graceful shutdown that ignores
// unfinished tasks or in-flight messages.
// This function returns a channel that receives an error if one occurred, or
// nil if it successfully and gracefully terminated.
// If gracePeriod is set to less than 0, the function waits indefinitely.
TerminateAsync(gracePeriod time.Duration) <-chan error
// IsRunning checks if the process was successfully started and not yet stopped.
// Returns true if running, otherwise false.
IsRunning() bool
// IsTerminates checks if the message-delivery process is terminated.
// Returns true if terminated, otherwise false.
IsTerminated() bool
// IsTerminating checks if the message-delivery process termination is ongoing.
// Returns true if the message message-delivery process is being terminated,
// but termination is not yet complete, otherwise false.
IsTerminating() bool
// SetTerminationNotificationListener adds a callback to listen for
// non-recoverable interruption events.
SetTerminationNotificationListener(listener TerminationNotificationListener)
}
MessageHandler is a callback that can be registered to receive messages asynchronously.
type MessageHandler func(inboundMessage message.InboundMessage)
MessagePublishReceiptListener is a listener that can be registered for the delivery receipt events.
type MessagePublishReceiptListener func(PublishReceipt)
MessagePublisher represents the shared functionality between all publisher instances.
type MessagePublisher interface {
// Extend LifecycleControl for various lifecycle management functionality.
LifecycleControl
}
MessagePublisherHealthCheck allows applications to check and listen for events that indicate when message publishers are ready to publish. This is often used to handle various back pressure schemes, such as reject on full, and allows publishing to stop until the publisher can begin accepting more messages.
type MessagePublisherHealthCheck interface {
// IsReady checks if the publisher can publish messages. Returns true if the
// publisher can publish messages, otherwise false if the publisher is prevented from
// sending messages (e.g., a full buffer or I/O problems).
IsReady() bool
// SetPublisherReadinessListener registers a listener to be called when the
// publisher can send messages. Typically, the listener is notified after a
// Publisher instance raises an error indicating that the outbound message
// buffer is full.
SetPublisherReadinessListener(listener PublisherReadinessListener)
// NotifyWhenReady makes a request to notify the application when the
// publisher is ready. This function triggers a readiness notification if one
// needs to be sent, otherwise the next readiness notification is
// processed.
NotifyWhenReady()
}
MessageReceiver represents the shared functionality between all MessageReceivers
type MessageReceiver interface {
// Extend LifecycleControl for various lifecycle management functionality
LifecycleControl
// AddSubscription will subscribe to another message source on a PubSub+ Broker to receive messages from.
// Will block until subscription is added. Accepts *resource.TopicSubscription instances as the subscription.
// Returns a solace/errors.*IllegalStateError if the service is not running.
// Returns a solace/errors.*IllegalArgumentError if unsupported Subscription type is passed.
// Returns nil if successful.
AddSubscription(subscription resource.Subscription) error
// RemoveSubscription will unsubscribe from a previously subscribed message source on a broker
// such that no more messages will be received from it.
// Will block until subscription is removed.
// Accepts *resource.TopicSubscription instances as the subscription.
// Returns an solace/errors.*IllegalStateError if the service is not running.
// Returns a solace/errors.*IllegalArgumentError if unsupported Subscription type is passed.
// Returns nil if successful.
RemoveSubscription(subscription resource.Subscription) error
// AddSubscriptionAsync will subscribe to another message source on a PubSub+ Broker to receive messages from.
// Will block until subscription is added. Accepts *resource.TopicSubscription instances as the subscription.
// Returns a solace/errors.*IllegalStateError if the service is not running.
// Returns a solace/errors.*IllegalArgumentError if unsupported Subscription type is passed.
// Returns nil if successful.
AddSubscriptionAsync(subscription resource.Subscription, listener SubscriptionChangeListener) error
// RemoveSubscriptionAsymc will unsubscribe from a previously subscribed message source on a broker
// such that no more messages will be received from it. Will block until subscription is removed.
// Accepts *resource.TopicSubscription instances as the subscription.
// Returns an solace/errors.*IllegalStateError if the service is not running.
// Returns a solace/errors.*IllegalArgumentError if unsupported Subscription type is passed.
// Returns nil if successful.
RemoveSubscriptionAsync(subscription resource.Subscription, listener SubscriptionChangeListener) error
}
MessageReplayError indicates
type MessageReplayError struct {
// contains filtered or unexported fields
}
func (err *MessageReplayError) Error() string
Error returns the error message.
func (err *MessageReplayError) Unwrap() error
Unwrap returns the wrapped error.
MessagingService represents a broker that provides a messaging service.
type MessagingService interface {
// Connect connects the messaging service.
// This function blocks until the connection attempt is completed.
// Returns nil if successful, otherwise an error containing failure details, which may be the following:
// - solace/errors.*PubSubPlusClientError - If a connection error occurs.
// - solace/errors.*IllegalStateError - If MessagingService has already been terminated.
Connect() error
// ConnectAsync connects the messaging service asynchronously.
// Returns a channel that receives an event when completed.
// Channel (chan) receives nil if successful, otherwise an error containing failure details.
// For more information, see MessagingService.Connect.
ConnectAsync() <-chan error
// ConnectAsyncWithCallback connects the messaging service asynchonously.
// When complete, the specified callback is called with nil if successful,
// otherwise an error if not successful. In both cases, the messaging service
// is passed as well.
ConnectAsyncWithCallback(callback func(MessagingService, error))
// CreateDirectMessagePublisherBuilder creates a DirectMessagePublisherBuilder
// that can be used to configure direct message publisher instances.
CreateDirectMessagePublisherBuilder() DirectMessagePublisherBuilder
// CreateDirectMessageReceiverBuilder creates a DirectMessageReceiverBuilder
// that can be used to configure direct message receiver instances.
CreateDirectMessageReceiverBuilder() DirectMessageReceiverBuilder
// CreatePersistentMessagePublisherBuilder creates a PersistentMessagePublisherBuilder
// that can be used to configure persistent message publisher instances.
CreatePersistentMessagePublisherBuilder() PersistentMessagePublisherBuilder
// CreatePersistentMessageReceiverBuilder creates a PersistentMessageReceiverBuilder
// that can be used to configure persistent message receiver instances.
CreatePersistentMessageReceiverBuilder() PersistentMessageReceiverBuilder
// MessageBuilder creates an OutboundMessageBuilder that can be
// used to build messages to send via a message publisher.
MessageBuilder() OutboundMessageBuilder
// Disconnect disconnects the messaging service.
// The messaging service must be connected to disconnect.
// This function blocks until the disconnection attempt is completed.
// Returns nil if successful, otherwise an error containing failure details.
// A disconnected messaging service may not be reconnected.
// Returns solace/errors.*IllegalStateError if it is not yet connected.
Disconnect() error
// DisconnectAsync disconnects the messaging service asynchronously.
// Returns a channel (chan) that receives an event when completed.
// The channel receives nil if successful, otherwise an error containing the failure details
// For more information, see MessagingService.Disconnect.
DisconnectAsync() <-chan error
// DisconnectAsyncWithCallback disconnects the messaging service asynchronously.
// When complete, the specified callback is called with nil if successful, otherwise
// an error if not successful.
DisconnectAsyncWithCallback(callback func(error))
// IsConnected determines if the messaging service is operational and if Connect was previously
// called successfully.
// Returns true if the messaging service is connected to a remote destination, otherwise false.
IsConnected() bool
// AddReconnectionListener adds a new reconnection listener to the messaging service.
// The reconnection listener is called when reconnection events occur.
// Returns an identifier that can be used to remove the listener using RemoveReconnectionListener.
AddReconnectionListener(listener ReconnectionListener) uint64
// AddReconnectionAttemptListener adds a listener to receive reconnection-attempt notifications.
// The reconnection listener is called when reconnection-attempt events occur.
// Returns an identifier that can be used to remove the listener using RemoveReconnectionAttemptListener.
AddReconnectionAttemptListener(listener ReconnectionAttemptListener) uint64
// RemoveReconnectionListener removes a listener from the messaging service with the specified identifier.
RemoveReconnectionListener(listenerID uint64)
// RemoveReconnectionAttemptListener removes a listener from the messaging service with the specified identifier.
RemoveReconnectionAttemptListener(listenerID uint64)
// AddServiceInterruptionListener adds a listener to receive non-recoverable, service-interruption events.
// Returns an identifier othat can be used to remove the listener using RemoveServiceInterruptionListener.
AddServiceInterruptionListener(listener ServiceInterruptionListener) uint64
// RemoveServiceInterruptionListener removes a service listener to receive non-recoverable,
// service-interruption events with the specified identifier.
RemoveServiceInterruptionListener(listenerID uint64)
// GetApplicationID retrieves the application identifier.
GetApplicationID() string
// Metrics returns the metrics for this MessagingService instance.
Metrics() metrics.APIMetrics
// Info returns the API Info for this MessagingService instance.
Info() metrics.APIInfo
// Updates the value of a modifiable service property once the service has been created.
// Modifiable service properties include:
// - solace/config.AuthenticationPropertySchemeOAuth2AccessToken,
// whose update will be applied during the next reconnection attempt.
// - solace/config.AuthenticationPropertySchemeOAuth2OIDCIDToken,
// whose update will be applied during the next reconnection attempt.
//
// Modification of a service property may occur instantly, or may occur during the next
// service reconnection.
// Modification of a service property during an ongoing service reconnection may apply
// to the next reconnection attempt.
// property (ServiceProperty): The name of the property to modify.
// value (interface{}): The new value of the property.
//
// - solace/errors.*IllegalArgumentError: If the specified property cannot
// - be modified.
// - solace/errors.*IllegalStateError: If the specified property cannot
// be modified in the current service state.
// - solace/errors.*NativeError: If other transport or communication related errors occur.
UpdateProperty(property config.ServiceProperty, value interface{}) error
}
MessagingServiceBuilder is used to configure and build MessagingService instances.
type MessagingServiceBuilder interface {
// Build creates MessagingService based on the provided configuration.
// Returns the built MessagingService instance, otherwise nil if an error occurred.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build() (messagingService MessagingService, err error)
// BuildWithApplicationID creates MessagingService based on the provided configuration
// using the specified application identifier as the applicationID.
// Returns the created MessagingService instance, otherwise nil if an error occurred.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
BuildWithApplicationID(applicationID string) (messagingService MessagingService, err error)
// FromConfigurationProvider sets the configuration based on the specified configuration provider.
// The following are built in configuration providers:
// - ServicePropertyMap - This can be used to set a ServiceProperty to a value programatically.
//
// The ServicePropertiesConfigurationProvider interface can also be implemented by a type
// to have it act as a configuration factory by implementing the following:
//
// func (type MyType) GetConfiguration() ServicePropertyMap {...}
//
// Any properties provided by the configuration provider are layered over top of any
// previously set properties, including those set by specifying various strategies.
FromConfigurationProvider(provider config.ServicePropertiesConfigurationProvider) MessagingServiceBuilder
// WithAuthenticationStrategy configures the resulting messaging service
// with the specified authentication configuration
WithAuthenticationStrategy(authenticationStrategy config.AuthenticationStrategy) MessagingServiceBuilder
// WithRetryStrategy configures the resulting messaging service
// with the specified retry strategy
WithConnectionRetryStrategy(retryStrategy config.RetryStrategy) MessagingServiceBuilder
// WithMessageCompression configures the resulting messaging service
// with the specified compression factor. The builder attempts to use
// the specified compression-level with the provided host and port. It fails
// to build if an an atempt is made to use compression on a non-secured and
// non-compressed port.
WithMessageCompression(compressionFactor int) MessagingServiceBuilder
// WithReconnectionRetryStrategy configures the resulting messaging service
// with the specified reconnection strategy.
WithReconnectionRetryStrategy(retryStrategy config.RetryStrategy) MessagingServiceBuilder
// WithTransportSecurityStrategy configures the resulting messaging service
// with the specified transport security strategy.
WithTransportSecurityStrategy(transportSecurityStrategy config.TransportSecurityStrategy) MessagingServiceBuilder
}
NativeError is a struct that stores the error message and subcode for the error message.
type NativeError struct {
// contains filtered or unexported fields
}
func NewNativeError(message string, subcode subcode.Code) *NativeError
NewNativeError returns a new solace.NativeError with the given message and subcode when applicable.
func (err *NativeError) Error() string
Error returns the error message from solace.NativeError.
func (err *NativeError) SubCode() subcode.Code
SubCode returns the subcode associated with the specified error, otherwise SubCodeNone if no subcode is relevant.
OutboundMessageBuilder allows construction of messages to be sent.
type OutboundMessageBuilder interface {
// Build creates an OutboundMessage instance based on the configured properties.
// Accepts additional configuration providers to apply only to the built message, with the
// last in the list taking precedence.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build(additionalConfiguration ...config.MessagePropertiesConfigurationProvider) (message.OutboundMessage, error)
// BuildWithByteArrayPayload creates a message with a byte array payload.
// Accepts additional configuration providers to apply only to the built message, with the
// last in the list taking precedence.
// Returns the built message, otherwise an error if one occurred.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
BuildWithByteArrayPayload(payload []byte, additionalConfiguration ...config.MessagePropertiesConfigurationProvider) (message message.OutboundMessage, err error)
// BuildWithStringPayload builds a new message with a string payload.
// Accepts additional configuration providers to apply only to the built message, with the
// last in the list taking precedence.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
BuildWithStringPayload(payload string, additionalConfiguration ...config.MessagePropertiesConfigurationProvider) (message message.OutboundMessage, err error)
// BuildWithMapPayload builds a new message with a SDTMap payload.
// Accepts additional configuration providers to apply only to the built message, with the
// last in the list taking precedence.
// If invalid data, ie. data not allowed as SDTData, is found in the
// map, this function will return a nil OutboundMessage and an error.
// Returns a solace/errors.*IllegalArgumentError if an invalid payload is specified.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
BuildWithMapPayload(payload sdt.Map, additionalConfiguration ...config.MessagePropertiesConfigurationProvider) (message message.OutboundMessage, err error)
// BuildWithStreamPayload builds a new message with a SDTStream payload.
// Accepts additional configuration providers to apply only to the built message, with the
// last in the list taking precedence.
// If invalid data, ie. data not allowed as SDTData, is found in the
// stream, this function returns a nil OutboundMessage and an error.
// Returns a solace/errors.*IllegalArgumentError if an invalid payload is specified.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
BuildWithStreamPayload(payload sdt.Stream, additionalConfiguration ...config.MessagePropertiesConfigurationProvider) (message message.OutboundMessage, err error)
// FromConfigurationProvider sets the given message properties to the resulting message.
// Both Solace defined config.MessageProperty keys as well as arbitrary user-defined
// property keys are accepted. If using custom defined properties, the date type can be
// any of sdt.Data supported types.
FromConfigurationProvider(properties config.MessagePropertiesConfigurationProvider) OutboundMessageBuilder
// WithProperty sets an individual message property on the resulting message.
// Both Solace defined config.MessageProperty keys as well as arbitrary user-defined
// property keys are accepted. If using custom defined properties, the date type can be
// any of sdt.Data supported types.
WithProperty(propertyName config.MessageProperty, propertyValue interface{}) OutboundMessageBuilder
// WithExpiration sets the message expiration time to the given time.
WithExpiration(t time.Time) OutboundMessageBuilder
// WithHTTPContentHeader sets the specified HTTP content-header on the message.
WithHTTPContentHeader(contentType, contentEncoding string) OutboundMessageBuilder
// WithPriority sets the priority of the message, where the priority is a value between 0 (lowest) and 255 (highest).
WithPriority(priority int) OutboundMessageBuilder
// WithApplicationMessageId sets the application message ID of the message. It is carried in the message metadata
// and is used for application to application signaling.
WithApplicationMessageID(messageID string) OutboundMessageBuilder
// WithApplicationMessageType sets the application message type for a message. It is carried in the message metadata
// and is used for application to application signaling.
WithApplicationMessageType(messageType string) OutboundMessageBuilder
// WithSequenceNumber sets the sequence number for the message. The sequence number is carried in the message metadata
// and is used for application to application signaling.
WithSequenceNumber(sequenceNumber uint64) OutboundMessageBuilder
// WithSenderID sets the sender ID for a message from a string. If config.ServicePropertyGenerateSenderID is enabled on
// the messaging service, then passing a string to this method will override the API generated sender ID.
WithSenderID(senderID string) OutboundMessageBuilder
// WithCorrelationID sets the correlation ID for the message. The correlation ID is user-defined and carried end-to-end.
// It can be matched in a selector, but otherwise is not relevant to the event broker. The correlation ID may be used
// for peer-to-peer message synchronization. In JMS applications, this field is carried as the JMSCorrelationID Message
// Header Field.
WithCorrelationID(correlationID string) OutboundMessageBuilder
}
PersistentMessagePublisher allows for the publishing of persistent messages (guaranteed messages).
type PersistentMessagePublisher interface {
MessagePublisher
MessagePublisherHealthCheck
// StartAsyncCallback starts the PersistentMessagePublisher asynchronously.
// Calls the callback when started with an error if one occurred, otherwise nil
// when successful.
StartAsyncCallback(callback func(PersistentMessagePublisher, error))
// SetPublishReceiptListener sets the listener to receive delivery receipts.
// PublishReceipt events are triggered once the API receives an acknowledgement
// when a message is received from the event broker.
// This should be set before the publisher is started to avoid dropping acknowledgements.
// The listener does not receive events from PublishAwaitAcknowledgement calls.
SetMessagePublishReceiptListener(listener MessagePublishReceiptListener)
// TerminateAsyncCallback terminates the PersistentMessagePublisher asynchronously.
// Calls the callback when terminated with nil if successful, otherwise an error if
// one occurred. When gracePeriod is a value less than 0, the function waits indefinitely.
TerminateAsyncCallback(gracePeriod time.Duration, callback func(error))
// PublishBytes sends a message of type byte array to the specified destination.
// Returns an error if one occurred while attempting to publish or if the publisher
// is not started/terminated. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners
// are called.
PublishBytes(message []byte, destination *resource.Topic) error
// PublishString sends a message of type string to the specified destination.
// Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners
// are called.
PublishString(message string, destination *resource.Topic) error
// Publish sends the specified message of type OutboundMessage built by a
// OutboundMessageBuilder to the specified destination.
// Optionally, you can provide properties in the form of OutboundMessageProperties to override
// any properties set on OutboundMessage. The properties argument can be nil to
// not set any properties.
// Optionally, provide a context that is available in the PublishReceiptListener
// registered with SetMessagePublishReceiptListener as GetUserContext.
// The context argument can be nil to not set a context. Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners
// are called.
Publish(message message.OutboundMessage, destination *resource.Topic, properties config.MessagePropertiesConfigurationProvider, context interface{}) error
// PublishAwaitAcknowledgement sends the specified message of type OutboundMessage
// and awaits a publish acknowledgement.
// Optionally, you can provide properties in the form of OutboundMessageProperties to override
// any properties set on OutboundMessage. The properties argument can be nil to
// not set any properties.
// If the specified timeout argument is less than 0, the function waits indefinitely.
// Possible errors include:
// - solace/errors.*PubSubPlusClientError - If the message could not be sent and all retry attempts failed.
// - solace/errors.*PublisherOverflowError - If messages are published faster than publisher's I/O
// capabilities allow. When publishing can be resumed, the registered PublisherReadinessListeners
// are called.
PublishAwaitAcknowledgement(message message.OutboundMessage, destination *resource.Topic, timeout time.Duration, properties config.MessagePropertiesConfigurationProvider) error
}
PersistentMessagePublisherBuilder allows for configuration of persistent message publisher instances.
type PersistentMessagePublisherBuilder interface {
// Build returns a new PersistentMessagePublisher based on the configured properties.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build() (messagePublisher PersistentMessagePublisher, err error)
// OnBackPressureReject sets the publisher back pressure strategy to reject
// where the publish attempts are rejected once the bufferSize (the number of messages), is reached.
// If bufferSize is 0, an error is thrown when the transport is full when attempting to publish.
// A buffer of the given size will be statically allocated when the publisher is built.
// Valid bufferSize is greater than or equal to 0.
OnBackPressureReject(bufferSize uint) PersistentMessagePublisherBuilder
// OnBackPressureWait sets the publisher back pressure strategy to wait where publish
// attempts block until there is space in the buffer of size bufferSize (the number of messages).
// A buffer of the given size will be statically allocated when the publisher is built.
// Valid bufferSize is greater than or equal to 1.
OnBackPressureWait(bufferSize uint) PersistentMessagePublisherBuilder
// FromConfigurationProvider configures the persistent publisher with the given properties.
// Built in PublisherPropertiesConfigurationProvider implementations include:
// - PublisherPropertyMap - A map of PublisherProperty keys to values.
FromConfigurationProvider(provider config.PublisherPropertiesConfigurationProvider) PersistentMessagePublisherBuilder
}
PersistentMessageReceiver allows for receiving persistent message (guaranteed messages).
type PersistentMessageReceiver interface {
MessageReceiver // Include all functionality of MessageReceiver.
// Ack acknowledges that a message was received.
Ack(message message.InboundMessage) error
// StartAsyncCallback starts the PersistentMessageReceiver asynchronously.
// Calls the callback when started with an error if one occurred, otherwise nil
// if successful.
StartAsyncCallback(callback func(PersistentMessageReceiver, error))
// TerminateAsyncCallback terminates the PersistentMessageReceiver asynchronously.
// Calls the callback when terminated with nil if successful, otherwise an error if
// one occurred. If gracePeriod is less than 0, the function waits indefinitely.
TerminateAsyncCallback(gracePeriod time.Duration, callback func(error))
// ReceiveAsync registers a callback to be called when new messages
// are received. Returns an error if one occurred while registering the callback.
// If a callback is already registered, it is replaced by the specified
// callback.
ReceiveAsync(callback MessageHandler) error
// ReceiveMessage receives a message synchronously from the receiver.
// Returns an error if the receiver is not started or already terminated.
// This function waits until the specified timeout to receive a message or waits
// forever if timeout value is negative. If a timeout occurs, a solace.TimeoutError
// is returned.
ReceiveMessage(timeout time.Duration) (message.InboundMessage, error)
// Pause pauses the receiver's message delivery to asynchronous message handlers.
// Pausing an already paused receiver has no effect.
// Returns an IllegalStateError if the receiver has not started or has already terminated.
Pause() error
// Resume unpause the receiver's message delivery to asynchronous message handlers.
// Resume a receiver that is not paused has no effect.
// Returns an IllegalStateError if the receiver has not started or has already terminated.
Resume() error
// ReceiverInfo returns a runtime accessor for the receiver information such as the remote
// resource to which it connects.
// Returns an IllegalStateError if the receiver has not started or has already terminated.
ReceiverInfo() (info PersistentReceiverInfo, err error)
}
PersistentMessageReceiverBuilder is used for configuration of PersistentMessageReceiver.
type PersistentMessageReceiverBuilder interface {
// Build creates a PersistentMessageReceiver with the specified properties.
// Returns solace/errors.*IllegalArgumentError if the queue is nil.
// Returns solace/errors.*InvalidConfigurationError if an invalid configuration is provided.
Build(queue *resource.Queue) (receiver PersistentMessageReceiver, err error)
// WithActivationPassivationSupport sets the listener to receiver broker notifications
// about state changes for the resulting receiver. This change can happen if there are
// multiple instances of the same receiver for high availability and activity is exchanged.
// This change is handled by the broker.
WithActivationPassivationSupport(listener ReceiverStateChangeListener) PersistentMessageReceiverBuilder
// WithMessageAutoAcknowledgement enables automatic acknowledgement on all receiver methods.
// Auto Acknowledgement will be performed after an acknowledgement after the receive callback
// is called when using ReceiveAsync, or after the message is passed to the application when
// using ReceiveMessage. In cases where underlying network connectivity fails, automatic
// acknowledgement processing is not guaranteed.
WithMessageAutoAcknowledgement() PersistentMessageReceiverBuilder
// WithMessageClientAcknowledgement disables automatic acknowledgement on all receiver methods
// and instead enables support for client acknowledgement for both synchronous and asynchronous
// message delivery functions. New persistent receiver builders default to client acknowledgement.
WithMessageClientAcknowledgement() PersistentMessageReceiverBuilder
// WithMessageSelector sets the message selector to the specified string.
// If an empty string is provided, the filter is cleared.
WithMessageSelector(filterSelectorExpression string) PersistentMessageReceiverBuilder
// WithMissingResourcesCreationStrategy sets the missing resource creation strategy
// defining what actions the API may take when missing resources are detected.
WithMissingResourcesCreationStrategy(strategy config.MissingResourcesCreationStrategy) PersistentMessageReceiverBuilder
// WithMessageReplay enables support for message replay using a specific replay
// strategy. Once started, the receiver replays using the specified strategy.
// Valid strategies include config.ReplayStrategyAllMessages(),
// config.ReplayStrategyTimeBased and config.ReplicationGroupMessageIDReplayStrategy.
WithMessageReplay(strategy config.ReplayStrategy) PersistentMessageReceiverBuilder
// WithSubscriptions sets a list of TopicSubscriptions to subscribe
// to when starting the receiver. Accepts *resource.TopicSubscription subscriptions.
WithSubscriptions(topics ...resource.Subscription) PersistentMessageReceiverBuilder
// FromConfigurationProvider configures the persistent receiver with the specified properties.
// The built-in ReceiverPropertiesConfigurationProvider implementations include:
// ReceiverPropertyMap, a map of ReceiverProperty keys to values
FromConfigurationProvider(provider config.ReceiverPropertiesConfigurationProvider) PersistentMessageReceiverBuilder
}
PersistentReceiverInfo provides information about the receiver at runtime.
type PersistentReceiverInfo interface {
// GetResourceInfo returns the remote endpoint (resource) information for the receiver.
GetResourceInfo() ResourceInfo
}
PublishFailureListener is a listener that can be registered for publish failure events.
type PublishFailureListener func(FailedPublishEvent)
PublishReceipt is the receipt for delivery of a persistent message.
type PublishReceipt interface {
// GetUserContext retrieves the context associated with the publish, if provided.
// Returns nil if no user context is set.
GetUserContext() interface{}
// GetTimeStamp retrieves the time. The time indicates when the event occurred, specifically the time when the
// acknowledgement was received by the API from the event broker, or if present, when the GetError error
// occurred.
GetTimeStamp() time.Time
// GetMessage returns an OutboundMessage that was successfully published.
GetMessage() message.OutboundMessage
// GetError retrieves an error if one occurred, which usually indicates a publish attempt failed.
// GetError returns nil on a successful publish, otherwise an error if a failure occurred
// while delivering the message.
GetError() error
// IsPersisted returns true if the event broker confirmed that the message was successfully received and persisted,
// otherwise false.
IsPersisted() bool
}
PublisherOverflowError indicates when publishing has stopped due to internal buffer limits. The pointer type *PublisherOverflowError is returned.
type PublisherOverflowError struct {
// contains filtered or unexported fields
}
func (err *PublisherOverflowError) Error() string
Error returns the error message.
func (err *PublisherOverflowError) Unwrap() error
Unwrap returns the wrapped error.
PublisherReadinessListener defines a function that can be registered to receive notifications from a publisher instance for readiness.
type PublisherReadinessListener func()
ReceiverState represents the various states the receiver can be in, such as Active or Passive. This property is controlled by the remote broker.
type ReceiverState byte
const (
// ReceiverActive is the state in which the receiver receives messages from a broker.
ReceiverActive ReceiverState = iota
// ReceiverPassive is the state in which the receiver would not receive messages from a broker.
// Often, this is because another instance of the receiver became active, so this instance became
// passive.
ReceiverPassive
)
ReceiverStateChangeListener is a listener that can be registered on a receiver to be notified of changes in receiver state by the remote broker.
type ReceiverStateChangeListener func(oldState ReceiverState, newState ReceiverState, timestamp time.Time)
ReconnectionAttemptListener is a handler that can be registered to a MessagingService. It is called when a session is disconnected and reconnection attempts have begun.
type ReconnectionAttemptListener func(event ServiceEvent)
ReconnectionListener is a handler that can be registered to a MessagingService. It is called when a session was disconnected and subsequently reconnected.
type ReconnectionListener func(event ServiceEvent)
ResourceInfo provides information about a resouce at runtime.
type ResourceInfo interface {
// GetName returns the name of the resource.
GetName() string
// IsDurable returns true is a resource is durable, otherwise false.
IsDurable() bool
}
ServiceEvent interface represents a messaging service event that applications can listen for.
type ServiceEvent interface {
// GetTimestamp retrieves the timestamp of the event.
GetTimestamp() time.Time
// GetBrokerURI retrieves the URI of the broker.
GetBrokerURI() string
// GetMessage retrieves the message contents.
GetMessage() string
// GetCause retrieves the cause of the client error.
GetCause() error
}
ServiceInterruptionListener is a handler that can be registered to a MessagingService. It is called when a session is disconncted and the connection is unrecoverable.
type ServiceInterruptionListener func(event ServiceEvent)
ServiceUnreachableError indicates that a remote service connection could not be established. The pointer type *ServiceUnreachableError is returned.
type ServiceUnreachableError struct {
// contains filtered or unexported fields
}
func (err *ServiceUnreachableError) Error() string
Error returns the error message.
func (err *ServiceUnreachableError) Unwrap() error
Unwrap returns the wrapped error.
SubscriptionChangeListener is a callback that can be set on async subscription operations that allows for handling of success or failure. The callback will be passed the subscription in question, the operation (either SubscriptionAdded or SubscriptionRemoved), and the error or nil if no error was thrown while adding the subscription.
type SubscriptionChangeListener func(subscription resource.Subscription, operation SubscriptionOperation, errOrNil error)
SubscriptionOperation represents the operation that triggered a SubscriptionChangeListener callback
type SubscriptionOperation byte
const (
// SubscriptionAdded is the resulting subscription operation from AddSubscriptionAsync
SubscriptionAdded SubscriptionOperation = iota
// SubscriptionRemoved is the resulting subscription operation from RemoveSubscription
SubscriptionRemoved
)
TerminationEvent represents a non-recoverable receiver or publisher unsolicited termination for which applications can listen.
type TerminationEvent interface {
// GetTimestamp retrieves the timestamp of the event.
GetTimestamp() time.Time
// GetMessage retrieves the event message.
GetMessage() string
// GetCause retrieves the cause of the client exception, if any.
// Returns the error event or nil if no cause is present.
GetCause() error
}
TerminationNotificationListener represents a listener that can be registered with a LifecycleControl instance to listen for termination events.
type TerminationNotificationListener func(TerminationEvent)
TimeoutError indicates that a timeout error occurred. The pointer type *TimeoutError is returned.
type TimeoutError struct {
// contains filtered or unexported fields
}
func (err *TimeoutError) Error() string
Error returns the error message.
func (err *TimeoutError) Unwrap() error
Unwrap returns the wrapped error.
| Name | Synopsis |
|---|---|
| config | Package config contains the following constructs used for configuration: |
| logging | Package logging allows for configuration of the API's logging levels. |
| message | Package message contains the type definitions of InboundMessage and OutboundMessage. |
| rgmid | Package rgmid contains the ReplicationGroupMessageID interface. |
| sdt | Package sdt contains the types needed to work with Structured Data on a message. |
| metrics | Package metrics contains the various metrics that can be retrieved as well as the interface for retrieving the metrics. |
| resource | Package resource contains types and factory functions for various broker resources such as topics and queues. |
| subcode | Package subcode contains the subcodes returned from the Solace PubSub+ Messaging API for C. The subcodes are generated in subcode_generated.go. |