Awesome

KubeMQ Java SDK

The KubeMQ SDK for Java enables Java developers to seamlessly communicate with the KubeMQ server, implementing various communication patterns such as Events, EventStore, Commands, Queries, and Queues.

• KubeMQ Java SDK
  • Prerequisites
  • Installation
  • Running Examples
  • Building from Source
  • SDK Overview
  • KubeMQ Client Configuration
    • Configuration Parameters
    • Example Usage
    • Notes
  • Optional Ping Operation
    • Ping Method
      • Return Value: ServerInfo
      • Example Usage
    • When to Use Ping
    • Important Notes
  • PubSub Events Operations
    • Create Channel
      • Request Parameters
      • Response
      • Example
    • Delete Channel
      • Request Parameters
      • Response
      • Example
    • List Channels
      • Request Parameters
      • Response
      • Example
    • Send Event Message
      • Request: EventMessage Class Attributes
      • Response
      • Example
    • Subscribe To Events Messages
      • Request: EventsSubscription Class Attributes
      • Response
      • Callback: EventMessageReceived Class Detail
      • Example
  • PubSub EventsStore Operations
    • Create Channel
      • Request Parameters
      • Response
      • Example
    • Delete Channel
      • Request Parameters
      • Response
      • Example
    • List Channels
      • Request Parameters
      • Response
      • Example
    • Send EventStore Message
      • Request: EventStoreMessage Class Attributes
      • Response
      • Example
    • Subscribe To EventsStore Messages
      • Request: EventsStoreSubscription Class Attributes
      • EventsStoreType Options
      • Response
      • Callback: EventStoreMessageReceived Class Detail
      • Example
  • Commands & Queries – Commands Operations
    • Create Channel
      • Request Parameters
      • Response
      • Example
    • Delete Channel
      • Request Parameters
      • Response
      • Example
    • List Channels
      • Request Parameters
      • Response
      • Example
    • Send Command Request
      • Request: CommandMessage Class Attributes
      • Response: CommandResponseMessage Class Attributes
      • Example
    • Subscribe To Commands
      • Request: CommandsSubscription Class Attributes
      • Response
      • Callback: CommandMessageReceived Class Attributes
      • Command Response: CommandResponseMessage Class Attributes
      • Example
  • Commands & Queries – Queries Operations
    • Create Channel
      • Request Parameters
      • Response
      • Example
    • Delete Channel
      • Request Parameters
      • Response
      • Example
    • List Channels
      • Request Parameters
      • Response
      • Example
    • Send Query Request
      • Request: QueryMessage Class Attributes
      • Response: QueryResponseMessage Class Attributes
      • Example
    • Subscribe To Queries
      • Request: QueriesSubscription Class Attributes
      • Response
      • Callback: QueryMessageReceived Class Attributes
      • Query Response: QueryResponseMessage Class Attributes
      • Example
  • Queues Operations
    • Create Channel
      • Request Parameters
      • Response
      • Example
    • Delete Channel
      • Request Parameters
      • Response
      • Example
    • List Channels
      • Request Parameters
      • Response
      • Example
    • Send Queue Message
      • Request: QueueMessage Class Attributes
      • Response: QueueSendResult Class Attributes
      • Example
    • Receive Queue Messages
      • Request: QueuesPollRequest Class Attributes
      • Response: QueuesPollResponse Class Attributes
      • Example
      • Message Handling Options:
      • Additional Example: Bulk Message Handling
    • Waiting Queue Messages
      • Request Parameters
      • Response: QueueMessagesWaiting Class Attributes
      • QueueMessageWaitingPulled Class Attributes
      • Example
      • Important Notes:
    • Pull Messages
      • Request Parameters
      • Response: QueueMessagesPulled Class Attributes
      • QueueMessageWaitingPulled Class Attributes
      • Example
      • Important Notes:

Prerequisites

• Java Development Kit (JDK) 8 or higher
• 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:

<dependency>
   <groupId>io.kubemq.sdk</groupId>
   <artifactId>kubemq-sdk-Java</artifactId>
   <version>LATEST</version>
</dependency>

To build with Gradle, add the dependency to your build.gradle file:

compile group: 'io.kubemq.sdk', name: 'kubemq-sdk-Java', version: '2.0.0'

Running Examples

The examples are standalone projects that showcase the usage of the SDK. 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 Queries
• io.kubemq.example.pubsub: Examples related to Events and EventStore
• io.kubemq.example.queues: Examples related to Queues

Building from Source

Once you check out the code from GitHub, you can build it using Maven:

mvn clean install

This command will run the tests and install the JAR file to your local Maven repository. To skip the tests, use the following command:

mvn clean install -D skipTests=true

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. To create any client instance, you need to use the respective builder with at least two mandatory parameters: address (KubeMQ server address) and clientId.

Configuration Parameters

The table below describes all available configuration parameters:

Example Usage

Here's an example of how to create a client instance (using PubSubClient as an example):

PubSubClient pubSubClient = PubSubClient.builder()
    .address("localhost:50000")
    .clientId("test-client-id")
    .authToken("your-auth-token")
    .tls(true)
    .tlsCertFile("/path/to/cert.pem")
    .tlsKeyFile("/path/to/key.pem")
    .caCertFile("/path/to/cacert.crt")
    .maxReceiveize(5 * 1024 * 1024)  // 5 MB
    .reconnectIntervalSeconds(10)
    .keepAlive(true)
    .pingIntervalInSeconds(30)
    .pingTimeoutInSeconds(10)
    .logLevel(Level.DEBUG)
    .build();

Replace PubSubClient with QueuesClient or CQClient to create instances of other client types. The configuration parameters remain the same for all client types.

Notes

• For secure connections, set tls to true and provide the paths to your TLS certificate and key files.
• Adjust maxReceiveize based on your expected message sizes to optimize performance.
• Fine-tune reconnectIntervalSeconds, keepAlive, pingIntervalInSeconds, and pingTimeoutInSeconds 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

ServerInfo ping() throws IOException

Return Value: ServerInfo

The ping() method returns a ServerInfo object with the following attributes:

Example Usage

try {
    ServerInfo serverInfo = client.ping();
    System.out.println("Successfully connected to KubeMQ server: " + serverInfo.getHost());
} catch (IOException e) {
    System.err.println("Failed to ping KubeMQ server: " + e.getMessage());
}

When to Use Ping

The ping operation is optional and should be used judiciously. Here are some appropriate scenarios for using ping:

1. Initial Connection Verification: You may use ping once after creating the client to verify the initial connection.

2. Troubleshooting: If you suspect connectivity issues, ping can help diagnose if the problem is with the connection to the KubeMQ server.

3. 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

Create a new Events channel.

Request Parameters

Response

Example

public void createEventsChannel(String eventChannelName) {
    try {
        boolean isChannelCreated = pubSubClient.createEventsChannel(eventChannelName);
        System.out.println("Events Channel created: " + isChannelCreated);
    } catch (RuntimeException e) {
        System.err.println("Failed to create events channel: " + e.getMessage());
    }
}

Delete Channel

Delete an existing Events channel.

Request Parameters

Response

Example

public void deleteEventsChannel(String eventChannelName) {
    try {
        boolean isChannelDeleted = pubSubClient.deleteEventsChannel(eventChannelName);
        System.out.println("Events Channel deleted: " + isChannelDeleted);
    } catch (RuntimeException e) {
        System.err.println("Failed to delete events channel: " + e.getMessage());
    }
}

List Channels

Retrieve a list of Events channels.

Request Parameters

Response

Returns a List<PubSubChannel> where each PubSubChannel has the following attributes:

Example

public void listEventsChannels(String searchQuery) {
    try {
        System.out.println("Listing Events Channels");
        List<PubSubChannel> eventChannels = pubSubClient.listEventsChannels(searchQuery);
        eventChannels.forEach(channel -> {
            System.out.println("Name: " + channel.getName() + 
                               " Channel Type: " + channel.getType() + 
                               " Is Active: " + channel.getIsActive());
        });
    } catch (RuntimeException e) {
        System.err.println("Failed to list event channels: " + e.getMessage());
    }
}

Send Event Message

Send a message to an Events channel.

Request: EventMessage Class Attributes

Response

This method doesn't return a value. Successful execution implies the message was sent.

Example

public void sendEventMessage() {
  try {
    String data = "Any data can be passed in byte, JSON or anything";
    Map<String, String> tags = new HashMap<>();
    tags.put("tag1", "kubemq");
    tags.put("tag2", "kubemq2");

    EventMessage eventMessage = EventMessage.builder()
            .id(UUID.randomUUID().toString())
            .channel(eventChannelName)
            .metadata("something you want to describe")
            .body(data.getBytes())
            .tags(tags)
            .build();

    pubSubClient.sendEventsMessage(eventMessage);
    System.out.println("Event message sent ");
  } catch (RuntimeException e) {
    System.err.println("Failed to send event message: " + e.getMessage());
  }
}

Subscribe To Events Messages

Subscribes to receive messages from an Events channel.

Request: EventsSubscription Class Attributes

Response

This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.

Callback: EventMessageReceived Class Detail

Example

public void subscribeToEvents() {
  try {
    // Consumer for handling received events
    Consumer<EventMessageReceived> onReceiveEventCallback = event -> {
      System.out.println("Received event:");
      System.out.println("ID: " + event.getId());
      System.out.println("Channel: " + event.getChannel());
      System.out.println("Metadata: " + event.getMetadata());
      System.out.println("Body: " + new String(event.getBody()));
      System.out.println("Tags: " + event.getTags());
    };


    // Consumer for handling errors
    Consumer<String> onErrorCallback = error -> {
      System.err.println("Error Received: " + error);
    };

    EventsSubscription subscription = EventsSubscription.builder()
            .channel(eventChannelName)
            .onReceiveEventCallback(onReceiveEventCallback)
            .onErrorCallback(onErrorCallback)
            .build();

    pubSubClient.subscribeToEvents(subscription);
    System.out.println("Events Subscribed");

    // Wait for 10 seconds and call the cancel subscription
    try{
      Thread.sleep(10 * 1000);
      subscription.cancel();
    } catch(Exception ex){

    }

  } catch (RuntimeException e) {
    System.err.println("Failed to subscribe to events: " + e.getMessage());
  }
}

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

Create a new EventsStore channel.

Request Parameters

Response

Example

public void createEventsStoreChannel(String eventStoreChannelName) {
    try {
        boolean isChannelCreated = pubSubClient.createEventsStoreChannel(eventStoreChannelName);
        System.out.println("EventsStore Channel created: " + isChannelCreated);
    } catch (RuntimeException e) {
        System.err.println("Failed to create events store channel: " + e.getMessage());
    }
}

Delete Channel

Delete an existing EventsStore channel.

Request Parameters

Response

Example

public void deleteEventsStoreChannel(String eventStoreChannelName) {
    try {
        boolean isChannelDeleted = pubSubClient.deleteEventsStoreChannel(eventStoreChannelName);
        System.out.println("EventsStore Channel deleted: " + isChannelDeleted);
    } catch (RuntimeException e) {
        System.err.println("Failed to delete events store channel: " + e.getMessage());
    }
}

List Channels

Retrieve a list of EventsStore channels.

Request Parameters

Response

Returns a List<PubSubChannel> where each PubSubChannel has the following attributes:

Example

public void listEventsStoreChannels(String searchQuery) {
    try {
        System.out.println("Listing EventsStore Channels");
        List<PubSubChannel> eventStoreChannels = pubSubClient.listEventsStoreChannels(searchQuery);
        eventStoreChannels.forEach(channel -> {
            System.out.println("Name: " + channel.getName() + 
                               " Channel Type: " + channel.getType() + 
                               " Is Active: " + channel.getIsActive());
        });
    } catch (RuntimeException e) {
        System.err.println("Failed to list events store channels: " + e.getMessage());
    }
}

Send EventStore Message

Send a message to an EventsStore channel.

Request: EventStoreMessage Class Attributes

Note: At least one of metadata, body, or tags is required.

Response

Returns an EventSendResult object (details not provided in the original content).

Example

public void sendEventStoreMessage(String eventStoreChannelName) {
    try {
        String data = "Sample event store data";
        Map<String, String> tags = new HashMap<>();
        tags.put("tag1", "value1");
        tags.put("tag2", "value2");

        EventStoreMessage eventStoreMessage = EventStoreMessage.builder()
                .id(UUID.randomUUID().toString())
                .channel(eventStoreChannelName)
                .metadata("Sample metadata")
                .body(data.getBytes())
                .tags(tags)
                .build();
        
        EventSendResult result = pubSubClient.sendEventsStoreMessage(eventStoreMessage);
        System.out.println("Event store message sent. Result: " + result);
    } catch (RuntimeException e) {
        System.err.println("Failed to send event store message: " + e.getMessage());
    }
}

Subscribe To EventsStore Messages

Subscribes to receive messages from an EventsStore channel.

Request: EventsStoreSubscription Class Attributes

EventsStoreType Options

Response

This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.

Callback: EventStoreMessageReceived Class Detail

Example

public void subscribeToEventsStore() {
  try {
    // Consumer for handling received event store messages
    Consumer<EventStoreMessageReceived> onReceiveEventCallback = event -> {
      System.out.println("Received event store:");
      System.out.println("ID: " + event.getId());
      System.out.println("Channel: " + event.getChannel());
      System.out.println("Metadata: " + event.getMetadata());
      System.out.println("Body: " + new String(event.getBody()));
      System.out.println("Tags: " + event.getTags());
    };

    // Consumer for handling errors
    Consumer<String> onErrorCallback = error -> {
      System.err.println("Error Received: " + error);
    };

    EventsStoreSubscription subscription = EventsStoreSubscription.builder()
            .channel(eventStoreChannelName)
            //.group("All IT Team")
            .eventsStoreType(EventsStoreType.StartAtTime)
            .eventsStoreStartTime(Instant.now().minus(1, ChronoUnit.HOURS))
            .onReceiveEventCallback(onReceiveEventCallback)
            .onErrorCallback(onErrorCallback)
            .build();

    pubSubClient.subscribeToEventsStore(subscription);
    System.out.println("EventsStore Subscribed");

    // Wait for 10 seconds and call the cancel subscription
    try{
      Thread.sleep(10 * 1000);
      subscription.cancel();
    }catch(Exception ex){}

  } catch (RuntimeException e) {
    System.err.println("Failed to subscribe to events store: " + e.getMessage());
  }
}

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

Create a new Command channel.

Request Parameters

Response

Example

public void createCommandsChannel(String channelName) {
    try {
        boolean isChannelCreated = cqClient.createCommandsChannel(channelName);
        System.out.println("Commands channel created: " + isChannelCreated);
    } catch (RuntimeException e) {
        System.err.println("Failed to create commands channel: " + e.getMessage());
    }
}

Delete Channel

Delete an existing Command channel.

Request Parameters

Response

Example

public void deleteCommandsChannel(String channelName) {
    try {
        boolean isChannelDeleted = cqClient.deleteCommandsChannel(channelName);
        System.out.println("Commands channel deleted: " + isChannelDeleted);
    } catch (RuntimeException e) {
        System.err.println("Failed to delete commands channel: " + e.getMessage());
    }
}

List Channels

Retrieve a list of Command channels.

Request Parameters

Response

Returns a List<CQChannel> where each CQChannel has the following attributes:

Example

public void listCommandsChannels(String searchString) {
    try {
        List<CQChannel> channels = cqClient.listCommandsChannels(searchString);
        System.out.println("Command Channels:");
        channels.forEach(channel -> {
            System.out.println("Name: " + channel.getName() + 
                               ", Type: " + channel.getType() + 
                               ", Active: " + channel.getIsActive());
        });
    } catch (RuntimeException e) {
        System.err.println("Failed to list command channels: " + e.getMessage());
    }
}

Send Command Request

Send a command request to a Command channel.

Request: CommandMessage Class Attributes

Response: CommandResponseMessage Class Attributes

Example

public void sendCommandRequest(String channel) {
    try {
        Map<String, String> tags = new HashMap<>();
        tags.put("tag1", "Command Message example");
        tags.put("tag2", "cq1");
        
        CommandMessage commandMessage = CommandMessage.builder()
                .channel(channel)
                .body("Test Command".getBytes())
                .metadata("Metadata add some extra information")
                .tags(tags)
                .timeoutInSeconds(20)
                .build();

        CommandResponseMessage response = cqClient.sendCommandRequest(commandMessage);
        System.out.println("Command Response: " + response);
    } catch (RuntimeException e) {
        System.err.println("Failed to send command request: " + e.getMessage());
    }
}

Subscribe To Commands

Subscribes to receive command messages from a Command channel.

Request: CommandsSubscription Class Attributes

Response

This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.

Callback: CommandMessageReceived Class Attributes

Command Response: CommandResponseMessage Class Attributes

When responding to a received command, you should construct a CommandResponseMessage with the following attributes:

Example

public void subscribeToCommands(String channel) {
    try {
        Consumer<CommandMessageReceived> onReceiveCommandCallback = receivedCommand -> {
            System.out.println("Received Command: " + new String(receivedCommand.getBody()));
            
            // Create a response message
            CommandResponseMessage response = CommandResponseMessage.builder()
                .commandReceived(receivedCommand)
                .clientId("responder-client-id")  // Set your client ID here
                .requestId(receivedCommand.getId())  // Use the received command's ID as the request ID
                .isExecuted(true)
                .timestamp(LocalDateTime.now())
                .error(null)  // Set an error message if execution failed
                .build();

            // Send the response
            cqClient.sendResponseMessage(response);
        };

        Consumer<String> onErrorCallback = errorMessage -> {
            System.err.println("Command Subscription Error: " + errorMessage);
        };

        CommandsSubscription subscription = CommandsSubscription.builder()
                .channel(channel)
                .onReceiveCommandCallback(onReceiveCommandCallback)
                .onErrorCallback(onErrorCallback)
                .build();

        cqClient.subscribeToCommands(subscription);
        System.out.println("Subscribed to Commands channel: " + channel);
        
        // To cancel the subscription later:
        // subscription.cancel();
    } catch (RuntimeException e) {
        System.err.println("Failed to subscribe to commands: " + e.getMessage());
    }
}

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

Create a new Query channel.

Request Parameters

Response

Example

public void createQueriesChannel(String channelName) {
    try {
        boolean isChannelCreated = cqClient.createQueriesChannel(channelName);
        System.out.println("Queries channel created: " + isChannelCreated);
    } catch (RuntimeException e) {
        System.err.println("Failed to create queries channel: " + e.getMessage());
    }
}

Delete Channel

Delete an existing Query channel.

Request Parameters

Response

Example

public void deleteQueriesChannel(String channelName) {
    try {
        boolean isChannelDeleted = cqClient.deleteQueriesChannel(channelName);
        System.out.println("Queries channel deleted: " + isChannelDeleted);
    } catch (RuntimeException e) {
        System.err.println("Failed to delete queries channel: " + e.getMessage());
    }
}

List Channels

Retrieve a list of Query channels.

Request Parameters

Response

Returns a List<CQChannel> where each CQChannel has the following attributes:

Example

public void listQueriesChannels(String searchString) {
    try {
        List<CQChannel> channels = cqClient.listQueriesChannels(searchString);
        System.out.println("Query Channels:");
        channels.forEach(channel -> {
            System.out.println("Name: " + channel.getName() + 
                               ", Type: " + channel.getType() + 
                               ", Active: " + channel.getIsActive());
        });
    } catch (RuntimeException e) {
        System.err.println("Failed to list query channels: " + e.getMessage());
    }
}

Send Query Request

Send a query request to a Query channel.

Request: QueryMessage Class Attributes

Response: QueryResponseMessage Class Attributes

Example

public void sendQueryRequest(String channel) {
    try {
        Map<String, String> tags = new HashMap<>();
        tags.put("tag1", "Query Message example");
        tags.put("tag2", "cq1");
        
        QueryMessage queryMessage = QueryMessage.builder()
                .channel(channel)
                .body("Test Query".getBytes())
                .metadata("Metadata add some extra information")
                .tags(tags)
                .timeoutInSeconds(20)
                .build();

        QueryResponseMessage response = cqClient.sendQueryRequest(queryMessage);
        System.out.println("Query Response: " + response);
    } catch (RuntimeException e) {
        System.err.println("Failed to send query request: " + e.getMessage());
    }
}

Subscribe To Queries

Subscribes to receive query messages from a Query channel.

Request: QueriesSubscription Class Attributes

Response

This method doesn't return a value. It sets up a subscription that will invoke the provided callbacks.

Callback: QueryMessageReceived Class Attributes

Query Response: QueryResponseMessage Class Attributes

When responding to a received query, you should construct a QueryResponseMessage with the following attributes:

Example

public void subscribeToQueries(String channel) {
    try {
        Consumer<QueryMessageReceived> onReceiveQueryCallback = receivedQuery -> {
            System.out.println("Received Query: " + new String(receivedQuery.getBody()));
            
            // Process the query and prepare a response
            String responseData = "Processed query result";
            
            // Create a response message
            QueryResponseMessage response = QueryResponseMessage.builder()
                .queryReceived(receivedQuery)
                .clientId("responder-client-id")  // Set your client ID here
                .requestId(receivedQuery.getId())  // Use the received query's ID as the request ID
                .executed(true)
                .timestamp(LocalDateTime.now())
                .metadata("Response metadata")
                .body(responseData.getBytes())
                .error(null)  // Set an error message if execution failed
                .build();

            // Send the response
            cqClient.sendQueryResponse(response);
        };

        Consumer<String> onErrorCallback = errorMessage -> {
            System.err.println("Query Subscription Error: " + errorMessage);
        };

        QueriesSubscription subscription = QueriesSubscription.builder()
                .channel(channel)
                .onReceiveQueryCallback(onReceiveQueryCallback)
                .onErrorCallback(onErrorCallback)
                .build();

        cqClient.subscribeToQueries(subscription);
        System.out.println("Subscribed to Queries channel: " + channel);
        
        // To cancel the subscription later:
        // subscription.cancel();
    } catch (RuntimeException e) {
        System.err.println("Failed to subscribe to queries: " + e.getMessage());
    }
}

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

Create a new Queue channel.

Request Parameters

Response

Example

public void createQueueChannel(String queueChannelName) {
    try {
        boolean isChannelCreated = queuesClient.createQueuesChannel(queueChannelName);
        System.out.println("Queue Channel created: " + isChannelCreated);
    } catch (RuntimeException e) {
        System.err.println("Failed to create queue channel: " + e.getMessage());
    }
}

Delete Channel

Delete an existing Queue channel.

Request Parameters

Response

Example

public void deleteQueueChannel(String queueChannelName) {
    try {
        boolean isChannelDeleted = queuesClient.deleteQueuesChannel(queueChannelName);
        System.out.println("Queue Channel deleted: " + isChannelDeleted);
    } catch (RuntimeException e) {
        System.err.println("Failed to delete queue channel: " + e.getMessage());
    }
}

List Channels

Retrieve a list of Queue channels.

Request Parameters

Response

Returns a List<QueuesChannel> where each QueuesChannel has the following attributes:

Example

public void listQueueChannels(String searchString) {
    try {
        List<QueuesChannel> channels = queuesClient.listQueuesChannels(searchString);
        System.out.println("Queue Channels:");
        for (QueuesChannel channel : channels) {
            System.out.println("Channel Name: " + channel.getName());
            System.out.println("Type: " + channel.getType());
            System.out.println("Last Activity: " + channel.getLastActivity());
            System.out.println("Is Active: " + channel.getIsActive());
            System.out.println("Incoming Stats: " + channel.getIncoming());
            System.out.println("Outgoing Stats: " + channel.getOutgoing());
            System.out.println();
        }
    } catch (RuntimeException e) {
        System.err.println("Failed to list queue channels: " + e.getMessage());
    }
}

Send Queue Message

Send a message to a Queue channel.

Request: QueueMessage Class Attributes

Response: QueueSendResult Class Attributes

Example

public void sendQueueMessage() {
  System.out.println("\n============================== Send Queue Message Started =============================\n");
  try {
    Map<String, String> tags = new HashMap<>();
    tags.put("tag1", "kubemq");
    tags.put("tag2", "kubemq2");
    QueueMessage message = QueueMessage.builder()
            .body("Sending data in queue message stream".getBytes())
            .channel(channelName)
            .metadata("Sample metadata")
            .id(UUID.randomUUID().toString())
            // Optional parameters
            .tags(tags)
            .delayInSeconds(1)
            .expirationInSeconds(3600)
            .attemptsBeforeDeadLetterQueue(3)
            .deadLetterQueue("dlq-" + channelName)
            .build();

    QueueSendResult sendResult = queuesClient.sendQueuesMessage(message);

    System.out.println("Message sent Response:");
    System.out.println("ID: " + sendResult.getId());
    System.out.println("Sent At: " + sendResult.getSentAt());
    System.out.println("Expired At: " + sendResult.getExpiredAt());
    System.out.println("Delayed To: " + sendResult.getDelayedTo());
    System.out.println("Is Error: " + sendResult.isError());
    if (sendResult.isError()) {
      System.out.println("Error: " + sendResult.getError());
    }
  } catch (RuntimeException e) {
    System.err.println("Failed to send queue message: " + e.getMessage());
  }

}

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

Receive messages from a Queue channel.

Request: QueuesPollRequest Class Attributes

Response: QueuesPollResponse Class Attributes

Response: QueueMessageReceived class attributes

Here's the requested Markdown table for the QueueMessageReceived class:

Example

public void receiveQueuesMessages(String channelName) {
    try {
        QueuesPollRequest queuesPollRequest = QueuesPollRequest.builder()
                .channel(channelName)
                .pollMaxMessages(1)
                .pollWaitTimeoutInSeconds(10)
                .build();

        QueuesPollResponse pollResponse = queuesClient.receiveQueuesMessages(queuesPollRequest);
        
        System.out.println("Received Message Response:");
        System.out.println("RefRequestId: " + pollResponse.getRefRequestId());
        System.out.println("ReceiverClientId: " + pollResponse.getReceiverClientId());
        System.out.println("TransactionId: " + pollResponse.getTransactionId());
        
        if (pollResponse.isError()) {
            System.out.println("Error: " + pollResponse.getError());
        } else {
            pollResponse.getMessages().forEach(msg -> {
                System.out.println("Message ID: " + msg.getId());
                System.out.println("Message Body: " + new String(msg.getBody()));
                
                // Message handling options:
                
                // 1. Acknowledge message (mark as processed)
                msg.ack();
                
                // 2. Reject message (won't be requeued)
                // msg.reject();
                
                // 3. Requeue message (send back to queue)
                // msg.reQueue(channelName);
            });
        }
    } catch (RuntimeException e) {
        System.err.println("Failed to receive queue messages: " + e.getMessage());
    }
}


public void receiveExampleWithVisibility() {
        System.out.println("\n============================== receiveExampleWithVisibility =============================\n");
       try {
            QueuesPollRequest queuesPollRequest = QueuesPollRequest.builder()
                    .channel(channelName)
                    .pollMaxMessages(1)  // Pull 10 messages at a time
                    .pollWaitTimeoutInSeconds(10)
                    .visibilitySeconds(5)
                    .autoAckMessages(false)
                    .build();

            QueuesPollResponse pollResponse = queuesClient.receiveQueuesMessages(queuesPollRequest);

            System.out.println("Received Message Response:"+" TransactionId: " + pollResponse.getTransactionId());
            System.out.println("RefRequestId: " + pollResponse.getRefRequestId()+" ReceiverClientId: " + pollResponse.getReceiverClientId());

            if (pollResponse.isError()) {
                System.out.println("Error: " + pollResponse.getError());
            } else {
                pollResponse.getMessages().forEach(msg -> {
                    try {
                        System.out.println("Message ID: " + msg.getId()+" Message Body: " + new String(msg.getBody()));
                        Thread.sleep(1000);
                        msg.ack();
                        System.out.println("Acknowledged to message");
                    } catch (InterruptedException ex) {
                        Logger.getLogger(ReceiveMessageWithVisibilityExample.class.getName()).log(Level.SEVERE, null, ex);
                    }
                });
            }

        } catch (RuntimeException e) {
            System.err.println("Failed to receive queue messages: " + e.getMessage());
        }
    }

    public void receiveExampleWithVisibilityExpired() {
        System.out.println("\n============================== receiveExampleWithVisibilityExpired =============================\n");
        try {
            QueuesPollRequest queuesPollRequest = QueuesPollRequest.builder()
                    .channel(channelName)
                    .pollMaxMessages(1)  // Pull 10 messages at a time
                    .pollWaitTimeoutInSeconds(10)
                    .visibilitySeconds(2)
                    .autoAckMessages(false)
                    .build();

            QueuesPollResponse pollResponse = queuesClient.receiveQueuesMessages(queuesPollRequest);

            System.out.println("Received Message Response:"+" TransactionId: " + pollResponse.getTransactionId());
            System.out.println("RefRequestId: " + pollResponse.getRefRequestId()+" ReceiverClientId: " + pollResponse.getReceiverClientId());

            if (pollResponse.isError()) {
                System.out.println("Error: " + pollResponse.getError());
            } else {
                pollResponse.getMessages().forEach(msg -> {
                    try {
                        System.out.println("Message ID: " + msg.getId()+" Message Body: " + new String(msg.getBody()));
                        Thread.sleep(3000);
                        msg.ack();
                        System.out.println("Acknowledged to message");
                    } catch (InterruptedException ex) {
                        Logger.getLogger(ReceiveMessageWithVisibilityExample.class.getName()).log(Level.SEVERE, null, ex);
                    }
                });
            }

        } catch (RuntimeException e) {
            System.err.println("Failed to receive queue messages: " + e.getMessage());
        }
    }

    public void receiveExampleWithVisibilityExtension() {
        System.out.println("\n============================== receiveExampleWithVisibilityExtension =============================\n");
        try {
            QueuesPollRequest queuesPollRequest = QueuesPollRequest.builder()
                    .channel(channelName)
                    .pollMaxMessages(1)  // Pull 10 messages at a time
                    .pollWaitTimeoutInSeconds(10)
                    .visibilitySeconds(3)
                    .autoAckMessages(false)
                    .build();

            QueuesPollResponse pollResponse = queuesClient.receiveQueuesMessages(queuesPollRequest);

            System.out.println("Received Message Response:");
            System.out.println("RefRequestId: " + pollResponse.getRefRequestId());
            System.out.println("ReceiverClientId: " + pollResponse.getReceiverClientId());
            System.out.println("TransactionId: " + pollResponse.getTransactionId());

            if (pollResponse.isError()) {
                System.out.println("Error: " + pollResponse.getError());
            } else {
                pollResponse.getMessages().forEach(msg -> {
                    try {
                        System.out.println("Message ID: " + msg.getId()+" Message Body: " + new String(msg.getBody()));
                        Thread.sleep(1000);
                        msg.extendVisibilityTimer(3);
                        Thread.sleep(2000);
                        msg.ack();
                        System.out.println("Acknowledged to message");
                    } catch (InterruptedException ex) {
                        Logger.getLogger(ReceiveMessageWithVisibilityExample.class.getName()).log(Level.SEVERE, null, ex);
                    }
                });
            }

        } catch (RuntimeException e) {
            System.err.println("Failed to receive queue messages: " + e.getMessage());
        }
    }

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:

1. Acknowledge (ack): Mark the message as processed and remove it from the queue.
2. Reject: Reject the message. It won't be requeued.
3. 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.

public void receiveAndBulkHandleQueueMessages(String channelName) {
  System.out.println("\n============================== Receive and Bulk Handle Queue Messages =============================\n");
  try {
    QueuesPollRequest queuesPollRequest = QueuesPollRequest.builder()
            .channel(channelName)
            .pollMaxMessages(10)  // Increased to receive multiple messages
            .pollWaitTimeoutInSeconds(15)
            .build();

    QueuesPollResponse pollResponse = queuesClient.receiveQueuesMessages(queuesPollRequest);

    System.out.println("Received Message Response:");
    System.out.println("RefRequestId: " + pollResponse.getRefRequestId());
    System.out.println("ReceiverClientId: " + pollResponse.getReceiverClientId());
    System.out.println("TransactionId: " + pollResponse.getTransactionId());

    if (pollResponse.isError()) {
      System.out.println("Error: " + pollResponse.getError());
    } else {
      int messageCount = pollResponse.getMessages().size();
      System.out.println("Received " + messageCount + " messages.");

      // Print details of received messages
      pollResponse.getMessages().forEach(msg -> {
        System.out.println("Message ID: " + msg.getId());
        System.out.println("Message Body: " + new String(msg.getBody()));
      });

      // Acknowledge all messages
      pollResponse.ackAll();
      System.out.println("Acknowledged all messages.");

      // Reject all messages
      // pollResponse.rejectAll();
      // System.out.println("Rejected all messages.");

      // Requeue all messages
      // pollResponse.reQueueAll(channelName);
      // System.out.println("Requeued all messages.");
    }

  } catch (RuntimeException e) {
    System.err.println("Failed to receive or handle queue messages: " + e.getMessage());
  }
}

This example showcases the following bulk operations:

1. ackAll(): Acknowledges all received messages, marking them as processed and removing them from the queue.
2. requeueAll(String channel): Requeues all received messages back to the specified channel for later processing.
3. 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

Response: QueueMessagesWaiting Class Attributes

QueueMessageWaitingPulled Class Attributes

Example

public void getWaitingMessages() {
    System.out.println("\n============================== getWaitingMessages Started =============================\n");
    try {
        String channelName = "mytest-channel";
        int maxNumberOfMessages = 1;
        int waitTimeSeconds = 10;

        QueueMessagesWaiting rcvMessages = queuesClient.waiting(channelName, maxNumberOfMessages, waitTimeSeconds);
        
        if (rcvMessages.isError()) {
            System.out.println("Error occurred: " + rcvMessages.getError());
            return;
        }
        
        System.out.println("Waiting Messages Count: " + rcvMessages.getMessages().size());
        
        for (QueueMessageWaitingPulled msg : rcvMessages.getMessages()) {
            System.out.println("Message ID: " + msg.getId());
            System.out.println("Channel: " + msg.getChannel());
            System.out.println("Metadata: " + msg.getMetadata());
            System.out.println("Body: " + new String(msg.getBody()));
            System.out.println("From Client ID: " + msg.getFromClientId());
            System.out.println("Tags: " + msg.getTags());
            System.out.println("Timestamp: " + msg.getTimestamp());
            System.out.println("Sequence: " + msg.getSequence());
            System.out.println("Receive Count: " + msg.getReceiveCount());
            System.out.println("Is Re-routed: " + msg.isReRouted());
            System.out.println("Re-route From Queue: " + msg.getReRouteFromQueue());
            System.out.println("Expired At: " + msg.getExpiredAt());
            System.out.println("Delayed To: " + msg.getDelayedTo());
            System.out.println("Receiver Client ID: " + msg.getReceiverClientId());
            System.out.println("--------------------");
        }
    } catch (RuntimeException e) {
        System.err.println("Failed to get waiting messages: " + e.getMessage());
    }
}

This method allows you to peek at messages waiting in a specified queue channel without removing them. It's particularly useful for:

Important Notes:

1. Monitoring queue depth and content.
2. Implementing custom logic based on the number or content of waiting messages.
3. 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

Response: QueueMessagesPulled Class Attributes

QueueMessageWaitingPulled Class Attributes

Example

public void getPullMessages() {
    System.out.println("\n============================== getPullMessages Started =============================\n");
    try {
        String channelName = "mytest-channel";
        int maxNumberOfMessages = 1;
        int waitTimeSeconds = 10;

        QueueMessagesPulled rcvMessages = queuesClient.pull(channelName, maxNumberOfMessages, waitTimeSeconds);

        if (rcvMessages.isError()) {
            System.out.println("Error occurred: " + rcvMessages.getError());
            return;
        }

        System.out.println("Pulled Messages Count: " + rcvMessages.getMessages().size());

        for (QueueMessageWaitingPulled msg : rcvMessages.getMessages()) {
            System.out.println("Message ID: " + msg.getId());
            System.out.println("Channel: " + msg.getChannel());
            System.out.println("Metadata: " + msg.getMetadata());
            System.out.println("Body: " + new String(msg.getBody()));
            System.out.println("From Client ID: " + msg.getFromClientId());
            System.out.println("Tags: " + msg.getTags());
            System.out.println("Timestamp: " + msg.getTimestamp());
            System.out.println("Sequence: " + msg.getSequence());
            System.out.println("Receive Count: " + msg.getReceiveCount());
            System.out.println("Is Re-routed: " + msg.isReRouted());
            System.out.println("Re-route From Queue: " + msg.getReRouteFromQueue());
            System.out.println("Expired At: " + msg.getExpiredAt());
            System.out.println("Delayed To: " + msg.getDelayedTo());
            System.out.println("Receiver Client ID: " + msg.getReceiverClientId());
            System.out.println("--------------------");
        }
    } catch (RuntimeException e) {
        System.err.println("Failed to pull messages: " + e.getMessage());
    }
}

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 an isError flag and an error 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.