Ack Range

package io.kubemq.example.queuesstream;

import io.kubemq.sdk.queues.*;
import java.util.UUID;

/**
 * AckRangeExample for Queues Stream
 *
 * Demonstrates selectively acknowledging specific messages using individual
 * message ack()/reject() calls. This allows fine-grained control over which
 * messages in a batch are acknowledged.
 */
public class AckRangeExample {
    private static final String ADDRESS = "localhost:50000";
    private static final String CLIENT_ID = "java-queues-ack-range-client";
    private static final String CHANNEL = "java-queues.ack-range";

    public static void main(String[] args) {
        // Create a client connected to the KubeMQ server
        try (QueuesClient client = QueuesClient.builder().address(ADDRESS).clientId(CLIENT_ID).build()) {
            // Create the queue channel
            client.createQueuesChannel(CHANNEL);

            // Send messages to the queue
            for (int i = 1; i <= 5; i++) {
                client.sendQueueMessage(QueueMessage.builder()
                        .channel(CHANNEL).body(("Range msg " + i).getBytes()).build());
            }

            // Poll for a batch of messages with manual ack
            QueuesPollResponse response = client.receiveQueueMessages(QueuesPollRequest.builder()
                    .channel(CHANNEL).pollMaxMessages(5).pollWaitTimeoutInSeconds(5).build());

            System.out.println("Received " + response.getMessages().size() + " messages.");

            // Selectively settle by sequence: ack even sequences, reject odd ones
            for (QueueMessageReceived msg : response.getMessages()) {
                long sequence = msg.getSequence();
                if (sequence % 2 == 0) {
                    msg.ack();
                    System.out.println("  Acked: seq=" + sequence);
                } else {
                    msg.reject();
                    System.out.println("  Rejected: seq=" + sequence);
                }
            }

            // Clean up rejected messages (receive and auto-ack)
            QueuesPollResponse cleanup = client.receiveQueueMessages(QueuesPollRequest.builder()
                    .channel(CHANNEL).pollMaxMessages(10).pollWaitTimeoutInSeconds(1).autoAckMessages(true).build());
            System.out.println("\nRemaining messages cleaned up: " + cleanup.getMessages().size());

            // Clean up resources
            client.deleteQueuesChannel(CHANNEL);
        } catch (Exception e) {
            System.err.println("Error: " + e.getMessage());
        }
    }
}