Module messaging.replayablechannel
tharmigan/messaging.replayablechannel
Overview
This package provides a replay mechanism for messages in a messaging system. It allows for the retrieval and reprocessing of messages that have already been sent, which can be useful for debugging, testing, or recovering from errors.
Components
Message
The message is a wrapper around the actual content being sent in the messaging system. It includes metadata such as the message ID, and any other relevant information that helps identify and process the message.
Message Context
The message context is an object that provides functions to retrieve and update message content and the associated properties and metadata. This context is used to manage the state of the message as it is processed.
Handlers
Processors
The processors are functions that accepts the message context and perform operations on the message. They can modify the content, add or update or remove properties and metadata, or perform any other necessary processing. The processors are assumed to be idempotent, meaning that they can be applied multiple times without changing the result beyond the initial application. The processors are executed sequentially in the order they are defined.
Destinations
The destinations are the final targets where the processed messages will be sent. Each destination can have its own processing logic, and they can be configured to handle different types of messages or perform different actions based on the message content. The destionations does not have to be idempotent, meaning that they can change the state of the system or perform actions that are not repeatable. The destinations are executed in parallel, allowing for efficient processing of multiple messages at once.
Channel
The channel is a collection of processors and destinations that are used to process messages. The channel is defined as two distinct flows:
- Source Flow: This flow can only contain processors which are executed sequentially. It is used to prepare the message before it is sent to the destinations.
- Destinations Flow: This flow can a single destination or multiple destinations, which are executed in parallel. It is used to send the processed message to the final targets. Additionally, a destination can be configured with processors that will be executed before the message is sent to the destination.
The channel provides two functions:
-
execute: This function takes a message content, create a message and message context and processes the message through the source flow, and then sends it to the destinations flow. -
replay: This function takes the message which is already processed in the channel and attempts to reprocess it through the source flow and send it to the destinations flow. This is useful for debugging or recovering from errors, as it allows you to reprocess messages that have already been sent with the knowledge of their previous state where it effectively skips the successful destinations.
Classes
messaging.replayablechannel: Channel
Channel is a collection of processors and destination that can process messages in a defined flow.
Constructor
Initializes a new instance of Channel with the provided processors and destination.
init (string name, *ChannelConfiguration config)- name string - The name of the channel.
- config *ChannelConfiguration - The configuration for the channel, which includes source flow and destinations flow.
getName
function getName() returns stringGet the name of the channel.
Return Type
- string - Returns the name of the channel
getFailureStore
function getFailureStore() returns MessageStore?Get the failure store of the channel.
Return Type
- MessageStore? - Returns the failure store of the channel, or a nil value if no failure store is defined.
replay
function replay(Message message) returns ExecutionResult|ExecutionErrorReplay the channel execution flow.
Parameters
- message Message - The message to replay process.
Return Type
- ExecutionResult|ExecutionError - Returns an error if the message could not be processed, otherwise returns the execution result.
execute
function execute(anydata content, string[] skipDestinations) returns ExecutionResult|ExecutionErrorDispatch a message to the channel for processing with the defined processors and destinations.
Parameters
- content anydata - The message content to be processed.
- skipDestinations string[] (default []) - An array of destination names to skip during execution.
Return Type
- ExecutionResult|ExecutionError - Returns the execution result or an error if the processing failed.
messaging.replayablechannel: MessageContext
MessageContext encapsulates the message and the relevant properties. Additionally, it provides methods to manipulate the message properties and metadata.
getId
function getId() returns stringGet the unique identifier of the message.
Return Type
- string - The unique identifier of the message.
toRecord
function toRecord() returns MessageGet the message as a record.
Return Type
- Message - A record version of the message.
getContent
function getContent() returns anydataGet the content of the message.
Return Type
- anydata - The content of the message, which can be of any data type.
getProperty
function getProperty(string key) returns anydataGet message property by key.
Parameters
- key string - The key of the property to retrieve.
Return Type
- anydata - The value of the property if it exists, otherwise panics.
setProperty
function setProperty(string key, anydata value)Set a property in the message.
Parameters
- key string - The key of the property to set.
- value anydata - The value to set for the property.
removeProperty
function removeProperty(string key) returns anydataRemove a property from the message.
Parameters
- key string - The key of the property to remove.
Return Type
- anydata - If the property exists, it is removed; otherwise, it panics.
hasProperty
Check if a property exists in the message.
Parameters
- key string - The key of the property to check.
Return Type
- boolean - Returns true if the property exists, otherwise false.
Annotations
messaging.replayablechannel: DestinationConfig
functionDestination configuration annotation.
messaging.replayablechannel: DestinationRouterConfig
functionDestination router configuration annotation.
messaging.replayablechannel: FilterConfig
functionFilter configuration annotation.
messaging.replayablechannel: ProcessorConfig
functionProcessor configuration annotation.
messaging.replayablechannel: ProcessorRouterConfig
functionProcessor router configuration annotation.
messaging.replayablechannel: TransformerConfig
functionTransformer configuration annotation.
Records
messaging.replayablechannel: ChannelConfiguration
Represent the configuration for a channel.
Fields
- Fields Included from *MessageFlow
- sourceFlow SourceFlow
- destinationsFlow DestinationsFlow
- failureStore? MessageStore - The store to use for storing messages that failed to process
- replayListenerConfig? ReplayListenerConfiguration - The configuration for replaying messages in the channel.
messaging.replayablechannel: ErrorDetails
Error details.
Fields
- message Message - the message associated with the error, which includes an ID and the message content.
messaging.replayablechannel: ErrorInfo
Error information type.
Fields
- message string - A descriptive error message.
- stackTrace string[] - An array of strings representing the stack trace of the error.
- detail map<anydata> - A map containing additional details about the error, which can include any data type.
messaging.replayablechannel: ExecutionResult
Channel execution result.
Fields
- message Message - The message that was processed.
- destinationResults map<any>(default {}) - A map of destination names to their respective results.
messaging.replayablechannel: HandlerConfiguration
Handler related configuration.
Fields
- name string - The name of the handler.
messaging.replayablechannel: Message
Message type represents a message with content, error information, metadata, and properties.
Fields
- idreadonly string - Unique identifier for the message.
- content anydata - The actual content of the message, which can be of any data type.
- errorMsg? string - Optional error message if an error occurred during processing.
- errorStackTrace? string[] - Optional stack trace of the error if available.
- errorDetails? map<anydata> - A map containing additional details about the error, which can include any data type.
- metadata MessageMetadata(default {}) - Metadata associated with the message, such as processors to skip.
- properties map<anydata>(default {}) - A map of additional properties associated with the message.
messaging.replayablechannel: MessageFlow
Represents the flow of the message in the channel, which includes the source flow and the destinations flow.
Fields
- sourceFlow SourceFlow - The source flow that processes the message context.
- destinationsFlow DestinationsFlow - The destinations flow that processes the message context and sends the message to one or more destinations.
messaging.replayablechannel: MessageMetadata
Message metadata.
Fields
- skipDestinations string[](default []) - An array of destination names that should be skipped when processing the message.
messaging.replayablechannel: ReplayListenerConfiguration
Represents the replay listener configuration for the channel.
Fields
- replayStore? MessageStore - The store which is listened to for replaying messages. If not set, the channel's failure store will be used.
- deadLetterStore? MessageStore - The store to use for storing messages that could not be processed after the maximum number of replay attempts.
Errors
messaging.replayablechannel: Error
Generic error type.
Union types
messaging.replayablechannel: Processor
Processor
Represents a processor that can be a filter, transformer, or processor and can be attached to a channel for processing messages. Processors should be idompotent i.e. repeating the execution should not change the outcome or the channel state.
messaging.replayablechannel: SourceFlow
SourceFlow
Represents a source flow that processes the message context. The pipline can consist a single processor or an array of processors. The source pipline will run in the configured order and the result of the last processor will be used as the message content for the next step in the channel flow.
messaging.replayablechannel: DestinationFlow
DestinationFlow
Represents a destination flow that processes the message context and sends the message to a destination. The flow can consist of a single destination or an array of processors followed by a destination. The processors in the flow will run in the configured order and the result of the last processor will be used as the message content for the destination.
messaging.replayablechannel: DestinationsFlow
DestinationsFlow
Represents the flow of destinations that can be used to send the message to multiple destinations. The flow can consist of a single destination router or a single destination flow, or an array of destination flows. The destination flows will run in parallel and the results will be collected.
Intersection types
messaging.replayablechannel: ExecutionError
ExecutionError
Execution error type for the channel execution.
Tuple types
messaging.replayablechannel: DestinationWithProcessors
DestinationWithProcessors
Represents a destination with processors that processes the message context and sends the message to a destination. The flow can consist of an array of processors followed by a destination. The processors in the flow will run in the configured order and the result of the last processor will be used as the message content for the destination.
Function types
messaging.replayablechannel: Transformer
function(MessageContext) returns (anydata|error)Transformer
Represents a transformer function that processes the message content and returns a modified message content.
messaging.replayablechannel: Filter
function(MessageContext) returns (boolean|error)Filter
Represents a filter function that checks the message context and returns a boolean indicating whether the message should be processed further.
messaging.replayablechannel: GenericProcessor
function(MessageContext) returns (error?)GenericProcessor
Represents a generic message processor that can process the message and return an error if the processing fails.
messaging.replayablechannel: ProcessorRouter
function(MessageContext) returns (Processor|error?)ProcessorRouter
Represents a processor router function that processes the message context and returns a processor. An error can be returned if the routing fails, or nil if it wants to stop processing the message.
messaging.replayablechannel: Destination
function(MessageContext) returns (any|error)Destination
Represents a destination function that processes the message context and returns a result or an error if it failed to send the message to the destination. Destinations are typically contains a sender or a writer that sends or writes the message to a specific destination.
messaging.replayablechannel: DestinationRouter
function(MessageContext) returns (DestinationFlow|error?)DestinationRouter
Represents a destination router function that processes the message context and returns a destination or an error if the routing fails, or nil if it wants to stop processing the message.