Java (Springboot)
The KubeMQ SDK for Spring Boot enables Spring Boot developers to seamlessly communicate with the KubeMQ server, implementing various communication patterns such as Events, EventStore, Commands, Queries, and Queues.
Sources
SDK
Examples
Prerequisites
Java Development Kit (JDK) 17 or higher
Spring Boot 3.x
Maven
KubeMQ server running locally or accessible over the network
Installation
The recommended way to use the SDK for Java in your project is to add it as a dependency in Maven:
To build with Gradle, add the dependency to your build.gradle
file:
Running Examples
The examples are standalone projects that showcase the usage of the SDK with Spring Boot. To run the examples, ensure you have a running instance of KubeMQ. Import the project into any IDE of your choice (e.g., IntelliJ, Eclipse, NetBeans). The example project contains three packages demonstrating different implementations:
io.kubemq.example.cq
: Examples related to Commands and Queriesio.kubemq.example.pubsub
: Examples related to Events and EventStoreio.kubemq.example.queues
: Examples related to Queues
Building from Source
Once you check out the code from GitHub, you can build it using Maven:
This command will run the tests and install the JAR file to your local Maven repository. To skip the tests, use the following command:
SDK Overview
The SDK implements all communication patterns available through the KubeMQ server:
PubSub
Events
EventStore
Commands & Queries (CQ)
Commands
Queries
Queues
KubeMQ Client Configuration
All KubeMQ clients (PubSubClient, QueuesClient, and CQClient) share the same configuration parameters. You can use Spring Boot @Autowired
To inject any client instance. In Spring Boot application project you have to specify the property in configuration files. Minimum two mandatory parameters: kubemq.client.address
(KubeMQ server address) and kubemq.client.clientId
you need to specify.
Configuration Example: -
application.properties
application.yml
Configuration Parameters
The table below describes all available configuration parameters:
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
address | String | The address of the KubeMQ server. | None | Yes |
clientId | String | The client ID used for authentication. | None | Yes |
authToken | String | The authorization token for secure communication. | None | No |
tls | boolean | Indicates if TLS (Transport Layer Security) is enabled. | false | No |
tlsCertFile | String | The path to the TLS certificate file. | None | No |
tlsKeyFile | String | The path to the TLS key file. | None | No |
caCertFile | String | The path to the CA certificate file. | None | No |
maxReceiveSize | int | The maximum size of the messages to receive (in bytes). | 104857600 (100MB) | No |
reconnectIntervalSeconds | int | The interval in seconds between reconnection attempts. | 1 | No |
keepAlive | boolean | Indicates if the connection should be kept alive. | false | No |
pingIntervalInSeconds | int | The interval in seconds between ping messages. | 60 | No |
pingTimeoutInSeconds | int | The timeout in seconds for ping messages. | 30 | No |
logLevel | Level | The logging level to use. | Level.INFO | No |
Spring Boot enable component scan
Enable component scan for package io.kubmq
this is base package for SDK, so spring can scan for beans and make it available for Auto-wiring.
Example Usage
Here's an example of how to Autowire client instance:
The parameters specified in configuration file remain the same for all client types.
Notes
For secure connections, set
tls
totrue
and provide the paths to your TLS certificate and key files.Adjust
maxReceiveSize
based on your expected message sizes to optimize performance.Fine-tune
reconnectIntervalSeconds
,keepAlive
,pingIntervalInSeconds
, andpingTimeoutInSeconds
based on your network conditions and requirements.Choose an appropriate
logLevel
for your development or production environment.
Remember to handle any exceptions that might be thrown during client creation, such as connection errors or invalid configuration parameters.
Optional Ping Operation
All KubeMQ clients (PubSubClient, QueuesClient, and CQClient) provide an optional ping()
method to verify connectivity with the KubeMQ server. This method is not required for normal operations and should be used sparingly.
Ping Method
Return Value: ServerInfo
ServerInfo
The ping()
method returns a ServerInfo
object with the following attributes:
Attribute | Type | Description |
---|---|---|
host | String | The host address of the KubeMQ server |
version | String | The version of the KubeMQ server |
serverStartTime | long | The start time of the server (in seconds since epoch) |
serverUpTimeSeconds | long | The uptime of the server in seconds |
Example Usage
When to Use Ping
The ping operation is optional and should be used judiciously. Here are some appropriate scenarios for using ping:
Initial Connection Verification: You may use ping once after creating the client to verify the initial connection.
Troubleshooting: If you suspect connectivity issues, ping can help diagnose if the problem is with the connection to the KubeMQ server.
Long Periods of Inactivity: In applications with long periods of inactivity, you might use ping to check if the connection is still alive before performing an operation.
Important Notes
Not Required for Regular Operations: The ping operation is not needed for regular message sending or receiving operations. The client handles connection management internally.
Performance Consideration: Excessive use of ping can introduce unnecessary network traffic and potential performance overhead.
Not a Guarantee: A successful ping doesn't guarantee that all server functionalities are working correctly. It only verifies basic connectivity.
Error Handling: Always handle potential IOException when using ping, as network issues can occur.
Remember, the KubeMQ client is designed to handle connection management efficiently. In most cases, you can rely on the client to maintain the connection without explicit ping operations.
PubSub Events Operations
Create Channel
Creates a new Events channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to create | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelCreated | boolean | Indicates if channel was created |
Example
Delete Channel
Deletes an existing Events channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to delete | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelDeleted | boolean | Indicates if channel was deleted |
Example
List Channels
Retrieves a list of Events channels.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
searchQuery | String | Search query to filter channels (optional) | None | No |
Response
Returns a List<PubSubChannel>
where each PubSubChannel
has the following attributes:
Name | Type | Description |
---|---|---|
name | String | The name of the Pub/Sub channel. |
type | String | The type of the Pub/Sub channel. |
lastActivity | long | The timestamp of the last activity on the channel, represented in milliseconds since epoch. |
isActive | boolean | Indicates whether the channel is active or not. |
incoming | PubSubStats | The statistics related to incoming messages for this channel. |
outgoing | PubSubStats | The statistics related to outgoing messages for this channel. |
Example
Send Event Message
Sends a message to an Events channel.
Request: EventMessage
Class Attributes
EventMessage
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
id | String | Unique identifier for the event message. | None | No |
channel | String | The channel to which the event message is sent. | None | Yes |
metadata | String | Metadata associated with the event message. | None | No |
body | byte[] | Body of the event message in bytes. | Empty byte array | No |
tags | Map<String, String> | Tags associated with the event message as key-value pairs. | Empty Map | No |
Response
This method doesn't return a value. Successful execution implies the message was sent.
Example
Subscribe To Events Messages
Subscribes to receive messages from an Events channel.
Request: EventsSubscription
Class Attributes
EventsSubscription
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channel | String | The channel to subscribe to. | None | Yes |
group | String | The group to subscribe with. | None | No |
onReceiveEventCallback | Consumer | Callback function to be called when an event message is received. | None | Yes |
onErrorCallback | Consumer | Callback function to be called when an error occurs. | None | No |
Response
This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.
Callback: EventMessageReceived
Class Detail
EventMessageReceived
Class DetailName | Type | Description |
---|---|---|
id | String | The unique identifier of the message. |
fromClientId | String | The ID of the client that sent the message. |
timestamp | long | The timestamp when the message was received, in seconds. |
channel | String | The channel to which the message belongs. |
metadata | String | The metadata associated with the message. |
body | byte[] | The body of the message. |
sequence | long | The sequence number of the message. |
tags | Map<String, String> | The tags associated with the message. |
Example
Note: Remember to handle the subscription lifecycle appropriately in your application. You may want to store the subscription object to cancel it when it's no longer needed.
PubSub EventsStore Operations
Create Channel
Creates a new EventsStore channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to create | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelCreated | boolean | Indicates if channel was created |
Example
Delete Channel
Deletes an existing EventsStore channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to delete | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelDeleted | boolean | Indicates if channel was deleted |
Example
List Channels
Retrieves a list of EventsStore channels.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
searchQuery | String | Search query to filter channels (optional) | None | No |
Response
Returns a List<PubSubChannel>
where each PubSubChannel
has the following attributes:
Name | Type | Description |
---|---|---|
name | String | The name of the Pub/Sub channel. |
type | String | The type of the Pub/Sub channel. |
lastActivity | long | The timestamp of the last activity on the channel, represented in milliseconds since epoch. |
isActive | boolean | Indicates whether the channel is active or not. |
incoming | PubSubStats | The statistics related to incoming messages for this channel. |
outgoing | PubSubStats | The statistics related to outgoing messages for this channel. |
Example
Send EventStore Message
Sends a message to an EventsStore channel.
Request: EventStoreMessage
Class Attributes
EventStoreMessage
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
id | String | Unique identifier for the event message. | None | No |
channel | String | The channel to which the event message is sent. | None | Yes |
metadata | String | Metadata associated with the event message. | None | No |
body | byte[] | Body of the event message in bytes. | Empty byte array | No |
tags | Map<String, String> | Tags associated with the event message as key-value pairs. | Empty Map | No |
Note: At least one of metadata
, body
, or tags
is required.
Response
Returns an EventSendResult
object (details not provided in the original content).
Example
Subscribe To EventsStore Messages
Subscribes to receive messages from an EventsStore channel.
Request: EventsStoreSubscription
Class Attributes
EventsStoreSubscription
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channel | String | The channel to subscribe to. | None | Yes |
group | String | The group to subscribe with. | None | No |
onReceiveEventCallback | Consumer | Callback function to be called when an event message is received. | None | Yes |
onErrorCallback | Consumer | Callback function to be called when an error occurs. | None | No |
eventsStoreType | EventsStoreType | Type of EventsStore subscription (e.g., StartAtTime, StartAtSequence) | None | Yes |
eventsStoreStartTime | Instant | Start time for EventsStore subscription (if applicable) | None | Conditional |
EventsStoreType Options
Type | Value | Description |
---|---|---|
Undefined | 0 | Default value, should be explicitly set to a valid type before use |
StartNewOnly | 1 | Start storing events from the point when the subscription is made |
StartFromFirst | 2 | Start storing events from the first event available |
StartFromLast | 3 | Start storing events from the last event available |
StartAtSequence | 4 | Start storing events from a specific sequence number |
StartAtTime | 5 | Start storing events from a specific point in time |
StartAtTimeDelta | 6 | Start storing events from a specific time delta in seconds |
Response
This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.
Callback: EventStoreMessageReceived
Class Detail
EventStoreMessageReceived
Class DetailName | Type | Description |
---|---|---|
id | String | The unique identifier of the message. |
fromClientId | String | The ID of the client that sent the message. |
timestamp | long | The timestamp when the message was received, in seconds. |
channel | String | The channel to which the message belongs. |
metadata | String | The metadata associated with the message. |
body | byte[] | The body of the message. |
sequence | long | The sequence number of the message. |
tags | Map<String, String> | The tags associated with the message. |
Example
Note: Remember to handle the subscription lifecycle appropriately in your application. You may want to store the subscription object to cancel it when it's no longer needed.
Commands & Queries – Commands Operations
Create Channel
Creates a new Command channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to create | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelCreated | boolean | Indicates if channel was created |
Example
Delete Channel
Deletes an existing Command channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to delete | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelDeleted | boolean | Indicates if channel was deleted |
Example
List Channels
Retrieves a list of Command channels.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
searchString | String | Search query to filter channels (optional) | None | No |
Response
Returns a List<CQChannel>
where each CQChannel
has the following attributes:
Name | Type | Description |
---|---|---|
name | String | The name of the channel. |
type | String | The type of the channel. |
lastActivity | long | The timestamp of the last activity on the channel. |
isActive | boolean | Indicates whether the channel is currently active. |
incoming | CQStats | Statistics about incoming messages to the channel. |
outgoing | CQStats | Statistics about outgoing messages from the channel. |
Example
Send Command Request
Sends a command request to a Command channel.
Request: CommandMessage
Class Attributes
CommandMessage
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
id | String | The ID of the command message. | None | Yes |
channel | String | The channel through which the command message will be sent. | None | Yes |
metadata | String | Additional metadata associated with the command message. | None | No |
body | byte[] | The body of the command message as bytes. | Empty byte array | No |
tags | Map<String, String> | A dictionary of key-value pairs representing tags associated with the command message. | Empty Map | No |
timeoutInSeconds | int | The maximum time in seconds for waiting to response. | None | Yes |
Response: CommandResponseMessage
Class Attributes
CommandResponseMessage
Class AttributesName | Type | Description |
---|---|---|
commandReceived | CommandMessageReceived | The command message received in the response. |
clientId | String | The client ID associated with the command response. |
requestId | String | The unique request ID of the command response. |
isExecuted | boolean | Indicates if the command has been executed. |
timestamp | LocalDateTime | The timestamp when the command response was created. |
error | String | The error message if there was an error. |
Example
Subscribe To Commands
Subscribes to receive command messages from a Command channel.
Request: CommandsSubscription
Class Attributes
CommandsSubscription
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channel | String | The channel for the subscription. | None | Yes |
group | String | The group associated with the subscription. | None | No |
onReceiveCommandCallback | Consumer | Callback function for receiving commands. | None | Yes |
onErrorCallback | Consumer | Callback function for error handling. | None | No |
Response
This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.
Callback: CommandMessageReceived
Class Attributes
CommandMessageReceived
Class AttributesName | Type | Description |
---|---|---|
id | String | The unique identifier of the command message. |
fromClientId | String | The ID of the client who sent the command message. |
timestamp | Instant | The timestamp when the command message was received. |
channel | String | The channel through which the command message was sent. |
metadata | String | Additional metadata associated with the command message. |
body | byte[] | The body of the command message as bytes. |
replyChannel | String | The channel to which the reply should be sent. |
tags | Map<String, String> | A dictionary of key-value pairs representing tags associated with the command message. |
Command Response: CommandResponseMessage
Class Attributes
CommandResponseMessage
Class AttributesWhen responding to a received command, you should construct a CommandResponseMessage
with the following attributes:
Name | Type | Description |
---|---|---|
commandReceived | CommandMessageReceived | The command message received in the response. |
clientId | String | The client ID associated with the command response. |
requestId | String | The unique request ID of the command response. |
isExecuted | boolean | Indicates if the command has been executed. |
timestamp | LocalDateTime | The timestamp when the command response was created. |
error | String | The error message if there was an error. |
Example
Note: Remember to handle the subscription lifecycle appropriately in your application. You may want to store the subscription object to cancel it when it's no longer needed. Also, ensure that you properly construct and send a CommandResponseMessage
for each received command to complete the request-response cycle.
Commands & Queries – Queries Operations
Create Channel
Creates a new Query channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to create | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelCreated | boolean | Indicates if channel was created |
Example
Delete Channel
Deletes an existing Query channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to delete | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelDeleted | boolean | Indicates if channel was deleted |
Example
List Channels
Retrieves a list of Query channels.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
searchString | String | Search query to filter channels (optional) | None | No |
Response
Returns a List<CQChannel>
where each CQChannel
has the following attributes:
Name | Type | Description |
---|---|---|
name | String | The name of the channel. |
type | String | The type of the channel. |
lastActivity | long | The timestamp of the last activity on the channel. |
isActive | boolean | Indicates whether the channel is currently active. |
incoming | CQStats | Statistics about incoming messages to the channel. |
outgoing | CQStats | Statistics about outgoing messages from the channel. |
Example
Send Query Request
Sends a query request to a Query channel.
Request: QueryMessage
Class Attributes
QueryMessage
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
id | String | The ID of the query message. | None | Yes |
channel | String | The channel through which the query message will be sent. | None | Yes |
metadata | String | Additional metadata associated with the query message. | None | No |
body | byte[] | The body of the query message as bytes. | Empty byte array | No |
tags | Map<String, String> | A dictionary of key-value pairs representing tags associated with the query message. | Empty Map | No |
timeoutInSeconds | int | The maximum time in seconds for waiting response. | None | Yes |
Response: QueryResponseMessage
Class Attributes
QueryResponseMessage
Class AttributesName | Type | Description |
---|---|---|
queryReceived | QueryMessageReceived | The query message received in the response. |
clientId | String | The client ID associated with the query response. |
requestId | String | The unique request ID of the query response. |
executed | boolean | Indicates if the query has been executed. |
timestamp | LocalDateTime | The timestamp when the query response was created. |
metadata | String | Additional metadata associated with the response. |
body | byte[] | The body of the query response as bytes. |
error | String | The error message if there was an error. |
Example
Subscribe To Queries
Subscribes to receive query messages from a Query channel.
Request: QueriesSubscription
Class Attributes
QueriesSubscription
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channel | String | The channel for the subscription. | None | Yes |
group | String | The group associated with the subscription. | None | No |
onReceiveQueryCallback | Consumer | Callback function for receiving queries. | None | Yes |
onErrorCallback | Consumer | Callback function for error handling. | None | No |
Response
This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.
Callback: QueryMessageReceived
Class Attributes
QueryMessageReceived
Class AttributesName | Type | Description |
---|---|---|
id | String | The unique identifier of the query message. |
fromClientId | String | The ID of the client who sent the query message. |
timestamp | Instant | The timestamp when the query message was received. |
channel | String | The channel through which the query message was sent. |
metadata | String | Additional metadata associated with the query message. |
body | byte[] | The body of the query message as bytes. |
replyChannel | String | The channel to which the reply should be sent. |
tags | Map<String, String> | A dictionary of key-value pairs representing tags associated with the query message. |
Query Response: QueryResponseMessage
Class Attributes
QueryResponseMessage
Class AttributesWhen responding to a received query, you should construct a QueryResponseMessage
with the following attributes:
Name | Type | Description |
---|---|---|
queryReceived | QueryMessageReceived | The query message received in the response. |
clientId | String | The client ID associated with the query response. |
requestId | String | The unique request ID of the query response. |
executed | boolean | Indicates if the query has been executed. |
timestamp | LocalDateTime | The timestamp when the query response was created. |
metadata | String | Additional metadata associated with the response. |
body | byte[] | The body of the query response as bytes. |
error | String | The error message if there was an error. |
Example
Note: Remember to handle the subscription lifecycle appropriately in your application. You may want to store the subscription object to cancel it when it's no longer needed. Also, ensure that you properly construct and send a QueryResponseMessage
for each received query to complete the request-response cycle.
Queues Operations
Create Channel
Creates a new Queue channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to create | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelCreated | boolean | Indicates if channel was created |
Example
Delete Channel
Deletes an existing Queue channel.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | Name of the channel you want to delete | None | Yes |
Response
Name | Type | Description |
---|---|---|
isChannelDeleted | boolean | Indicates if channel was deleted |
Example
List Channels
Retrieves a list of Queue channels.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
searchString | String | Search query to filter channels (optional) | None | No |
Response
Returns a List<QueuesChannel>
where each QueuesChannel
has the following attributes:
Name | Type | Description |
---|---|---|
name | String | The name of the queue channel. |
type | String | The type of the queue channel. |
lastActivity | long | The timestamp of the last activity in the queue channel. |
isActive | boolean | Indicates whether the queue channel is currently active. |
incoming | QueuesStats | The statistics for incoming messages in the queue channel. |
outgoing | QueuesStats | The statistics for outgoing messages in the queue channel. |
Example
Send Queue Message
Sends a message to a Queue channel.
Request: QueueMessage
Class Attributes
QueueMessage
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
id | String | The unique identifier for the message. | None | No |
channel | String | The channel of the message. | None | Yes |
metadata | String | The metadata associated with the message. | None | No |
body | byte[] | The body of the message. | new byte[0] | No |
tags | Map<String, String> | The tags associated with the message. | new HashMap<>() | No |
delayInSeconds | int | The delay in seconds before the message becomes available in the queue. | None | No |
expirationInSeconds | int | The expiration time in seconds for the message. | None | No |
attemptsBeforeDeadLetterQueue | int | The number of receive attempts allowed for the message before it is moved to the dead letter queue. | None | No |
deadLetterQueue | String | The dead letter queue where the message will be moved after reaching the maximum receive attempts. | None | No |
Response: QueueSendResult
Class Attributes
QueueSendResult
Class AttributesName | Type | Description |
---|---|---|
id | String | The unique identifier of the message. |
sentAt | LocalDateTime | The timestamp when the message was sent. |
expiredAt | LocalDateTime | The timestamp when the message will expire. |
delayedTo | LocalDateTime | The timestamp when the message will be delivered. |
isError | boolean | Indicates if there was an error while sending the message. |
error | String | The error message if |
Example
This method allows you to send a message to a specified Queue channel. You can customize various aspects of the message, such as its content, metadata, tags, delay, expiration, and dead letter queue settings. The response provides information about the sent message, including its ID, timestamps, and any potential errors.
Receive Queue Messages
Receives messages from a Queue channel.
Request: QueuesPollRequest
Class Attributes
QueuesPollRequest
Class AttributesName | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channel | String | The channel to poll messages from. | None | Yes |
pollMaxMessages | int | The maximum number of messages to poll. | 1 | No |
pollWaitTimeoutInSeconds | int | The wait timeout in seconds for polling messages. | 60 | No |
autoAckMessages | boolean | Indicates if messages should be auto-acknowledged. | false | No |
visibilitySeconds | int | Add a visibility timeout feature for messages. | 0 | No |
Response: QueuesPollResponse
Class Attributes
QueuesPollResponse
Class AttributesName | Type | Description |
---|---|---|
refRequestId | String | The reference ID of the request. |
transactionId | String | The unique identifier for the transaction. |
messages | List | The list of received queue messages. |
error | String | The error message, if any error occurred. |
isError | boolean | Indicates if there was an error. |
isTransactionCompleted | boolean | Indicates if the transaction is completed. |
activeOffsets | List | The list of active offsets. |
receiverClientId | String | The client ID of the receiver. |
visibilitySeconds | int | The visibility timeout for the message in seconds. |
isAutoAcked | boolean | Indicates whether the message was auto-acknowledged. |
Response: QueueMessageReceived
class attributes
Here's the requested Markdown table for the QueueMessageReceived
class:
Name | Type | Description |
---|---|---|
id | String | The unique identifier for the message. |
channel | String | The channel from which the message was received. |
metadata | String | Metadata associated with the message. |
body | byte[] | The body of the message in byte array format. |
fromClientId | String | The ID of the client that sent the message. |
tags | Map | Key-value pairs representing tags for the message. |
timestamp | Instant | The timestamp when the message was created. |
sequence | long | The sequence number of the message. |
receiveCount | int | The number of times the message has been received. |
isReRouted | boolean | Indicates whether the message was rerouted. |
reRouteFromQueue | String | The name of the queue from which the message was rerouted. |
expiredAt | Instant | The expiration time of the message, if applicable. |
delayedTo | Instant | The time the message is delayed until, if applicable. |
transactionId | String | The transaction ID associated with the message. |
isTransactionCompleted | boolean | Indicates whether the transaction for the message is completed. |
responseHandler | StreamObserver | The response handler for processing downstream requests. |
receiverClientId | String | The ID of the client receiving the message. |
visibilitySeconds | int | The visibility timeout for the message in seconds. |
isAutoAcked | boolean | Indicates whether the message was auto-acknowledged. |
Example
This method allows you to receive messages from a specified Queue channel. You can configure the polling behavior, including the maximum number of messages to receive and the wait timeout. The response provides detailed information about the received messages and the transaction.
Message Handling Options:
Acknowledge (ack): Mark the message as processed and remove it from the queue.
Reject: Reject the message. It won't be requeued.
Requeue: Send the message back to the queue for later processing.
Choose the appropriate handling option based on your application's logic and requirements.
Additional Example: Bulk Message Handling
This example demonstrates how to use the bulk operations ackAll
, rejectAll
, and requeueAll
on the QueuesPollResponse
object.
This example showcases the following bulk operations:
ackAll(): Acknowledges all received messages, marking them as processed and removing them from the queue.
requeueAll(String channel): Requeues all received messages back to the specified channel for later processing.
rejectAll(): Rejects all received messages. They won't be requeued.
These bulk operations are particularly useful when you need to apply the same action to all received messages based on certain conditions or business logic. They can significantly simplify your code when dealing with multiple messages at once.
Waiting Queue Messages
The "Waiting" operation allows you to retrieve information about messages waiting in a queue without removing them from the queue. This can be useful for monitoring queue status or implementing custom processing logic based on waiting messages.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | The name of the channel to check for waiting messages. | None | Yes |
maxNumberOfMessages | int | The maximum number of waiting messages to retrieve. | None | Yes |
waitTimeSeconds | int | The maximum time to wait for messages, in seconds. | None | Yes |
Response: QueueMessagesWaiting
Class Attributes
QueueMessagesWaiting
Class AttributesName | Type | Description |
---|---|---|
messages | List | List of waiting messages in the queue. |
isError | boolean | Indicates if there was an error. |
error | String | The error message, if any. |
QueueMessageWaitingPulled
Class Attributes
QueueMessageWaitingPulled
Class AttributesName | Type | Description |
---|---|---|
id | String | The unique identifier of the message. |
channel | String | The channel name of the message. |
metadata | String | Additional metadata associated with the message. |
body | byte[] | The body content of the message. |
fromClientId | String | The ID of the client that sent the message. |
tags | Map<String, String> | Key-value pairs of tags associated with the message. |
timestamp | Instant | The timestamp when the message was sent. |
sequence | long | The sequence number of the message in the queue. |
receiveCount | int | The number of times this message has been received. |
isReRouted | boolean | Indicates if the message has been re-routed. |
reRouteFromQueue | String | The original queue name if the message was re-routed. |
expiredAt | Instant | The timestamp when the message will expire. |
delayedTo | Instant | The timestamp until which the message is delayed for processing. |
receiverClientId | String | The ID of the client receiving the message. |
Example
This method allows you to peek at messages waiting in a specified queue channel without removing them. It's particularly useful for:
Important Notes:
Monitoring queue depth and content.
Implementing custom logic based on the number or content of waiting messages.
Previewing messages before deciding whether to process them.
Pull Messages
The "Pull Messages" operation allows you to retrieve and remove messages from a queue. Unlike the "Waiting" operation, this actually dequeues the messages, making them unavailable for other consumers.
Request Parameters
Name | Type | Description | Default Value | Mandatory |
---|---|---|---|---|
channelName | String | The name of the channel to pull messages from. | None | Yes |
maxNumberOfMessages | int | The maximum number of messages to pull. | None | Yes |
waitTimeSeconds | int | The maximum time to wait for messages, in seconds. | None | Yes |
Response: QueueMessagesPulled
Class Attributes
QueueMessagesPulled
Class AttributesName | Type | Description |
---|---|---|
messages | List | List of pulled messages from the queue. |
isError | boolean | Indicates if there was an error. |
error | String | The error message, if any. |
QueueMessageWaitingPulled
Class Attributes
QueueMessageWaitingPulled
Class AttributesName | Type | Description |
---|---|---|
id | String | The unique identifier of the message. |
channel | String | The channel name of the message. |
metadata | String | Additional metadata associated with the message. |
body | byte[] | The body content of the message. |
fromClientId | String | The ID of the client that sent the message. |
tags | Map<String, String> | Key-value pairs of tags associated with the message. |
timestamp | Instant | The timestamp when the message was sent. |
sequence | long | The sequence number of the message in the queue. |
receiveCount | int | The number of times this message has been received. |
isReRouted | boolean | Indicates if the message has been re-routed. |
reRouteFromQueue | String | The original queue name if the message was re-routed. |
expiredAt | Instant | The timestamp when the message will expire. |
delayedTo | Instant | The timestamp until which the message is delayed for processing. |
receiverClientId | String | The ID of the client receiving the message. |
Example
This example demonstrates how to pull messages from a specified queue channel, process them, and access all the available metadata for each message.
Important Notes:
The
pull
operation removes messages from the queue. Once pulled, these messages are no longer available to other consumers.The
QueueMessagesPulled
object includes anisError
flag and anerror
string, which should be checked before processing the messages.The structure of
QueueMessageWaitingPulled
is the same for both "waiting" and "pull" operations, providing consistent access to message metadata.
Last updated