MockNatsConnection

The MockNatsConnection class is a subclass of NatsConnection and is packaged in the current module with package scope. It is designed to have the necessary access for its functionality.

MessageQueueBenchmark

The MessageQueueBenchmark class is a public class used for benchmarking message queues. It is primarily used by software engineers to measure the performance of different message queue implementations. The class is responsible for setting up and executing benchmark tests on message queues, allowing developers to compare the speed and efficiency of various queueing systems.

public static void main(String[] args) throws InterruptedException {
    int msgCount = 10_000_000;
    NatsMessage[] msgs = new NatsMessage[msgCount];
    long start, end;
    System.out.printf("Running benchmarks with %s messages.\n", NumberFormat.getInstance().format(msgCount));
    System.out.println("Warmed up ...");
    byte[] warmBytes = "a".getBytes();
    MessageQueue warm = new MessageQueue(false);
    for (int j = 0; j < msgCount; j++) {
        msgs[j] = new ProtocolMessage(warmBytes);
        warm.push(msgs[j]);
    }
    System.out.println("Starting tests ...");
    MessageQueue push = new MessageQueue(false);
    start = System.nanoTime();
    for (int i = 0; i < msgCount; i++) {
        push.push(msgs[i]);
    }
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform %s push operations was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
    start = System.nanoTime();
    for (int i = 0; i < msgCount; i++) {
        push.popNow();
    }
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform %s popnow operations was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
    MessageQueue accumulateQueue = new MessageQueue(true);
    for (int j = 0; j < msgCount; j++) {
        msgs[j].next = null;
    }
    for (int i = 0; i < msgCount; i++) {
        accumulateQueue.push(msgs[i]);
    }
    start = System.nanoTime();
    while (accumulateQueue.length() > 0) {
        // works for single thread, but not multi
        accumulateQueue.accumulate(10_000, 100, Duration.ofMillis(500));
    }
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform accumulate %s messages was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
    for (int j = 0; j < msgCount; j++) {
        msgs[j].next = null;
    }
    final MessageQueue pushPopThreadQueue = new MessageQueue(false);
    final Duration timeout = Duration.ofMillis(10);
    final CompletableFuture<Void> go = new CompletableFuture<>();
    Thread pusher = new Thread(() -> {
        try {
            go.get();
            for (int i = 0; i < msgCount; i++) {
                pushPopThreadQueue.push(msgs[i]);
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    pusher.start();
    Thread popper = new Thread(() -> {
        try {
            go.get();
            for (int i = 0; i < msgCount; i++) {
                pushPopThreadQueue.pop(timeout);
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    popper.start();
    start = System.nanoTime();
    go.complete(null);
    pusher.join();
    popper.join();
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform %s pushes in one thread and pop with timeout in another was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
    final CompletableFuture<Void> go2 = new CompletableFuture<>();
    for (int j = 0; j < msgCount; j++) {
        msgs[j].next = null;
    }
    final MessageQueue pushPopNowThreadQueue = new MessageQueue(false);
    pusher = new Thread(() -> {
        try {
            go2.get();
            for (int i = 0; i < msgCount; i++) {
                pushPopNowThreadQueue.push(msgs[i]);
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    pusher.start();
    popper = new Thread(() -> {
        try {
            go2.get();
            for (int i = 0; i < msgCount; i++) {
                pushPopNowThreadQueue.popNow();
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    popper.start();
    start = System.nanoTime();
    go2.complete(null);
    pusher.join();
    popper.join();
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform %s pushes in one thread and pop nows in another was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
    final CompletableFuture<Void> go3 = new CompletableFuture<>();
    for (int j = 0; j < msgCount; j++) {
        msgs[j].next = null;
    }
    final MessageQueue pushAccumulateThreadQueue = new MessageQueue(true);
    pusher = new Thread(() -> {
        try {
            go3.get();
            for (int i = 0; i < msgCount; i++) {
                pushAccumulateThreadQueue.push(msgs[i]);
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    pusher.start();
    popper = new Thread(() -> {
        try {
            go3.get();
            int remaining = msgCount;
            while (remaining > 0) {
                NatsMessage cursor = pushAccumulateThreadQueue.accumulate(10_000, 100, Duration.ofMillis(500));
                while (cursor != null) {
                    remaining--;
                    cursor = cursor.next;
                }
            }
        } catch (Exception exp) {
            exp.printStackTrace();
        }
    });
    popper.start();
    start = System.nanoTime();
    go3.complete(null);
    pusher.join();
    popper.join();
    end = System.nanoTime();
    System.out.printf("\nTotal time to perform %s pushes in one thread and accumlates in another was %s ms, %s ns/op\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), NumberFormat.getInstance().format(((double) (end - start)) / ((double) (msgCount))));
    System.out.printf("\tor %s op/s\n", NumberFormat.getInstance().format(1_000_000_000L * ((double) (msgCount)) / ((double) (end - start))));
}

The main method in the io.nats.client.impl.MessageQueueBenchmark class performs a benchmark test on the MessageQueue class.

1. It initializes a variable msgCount with a value of 10 million and creates an array msgs of NatsMessage objects with the same size.
2. It prints the number of messages being used in the benchmark test.
3. It prints "Warmed up ..." to indicate that the warm-up phase is starting.
4. It creates a MessageQueue called warm and initializes it with 10 million ProtocolMessage objects containing a byte array with the letter 'a'.
5. It prints "Starting tests ..." to indicate that the actual benchmark tests are starting.
6. It creates a new MessageQueue called push.
7. It measures the time it takes to push all the msgs into the push queue and prints the total time and operations per second for the push operations.
8. It measures the time it takes to pop all the messages from the push queue and prints the total time and operations per second for the pop operations.
9. It creates a new MessageQueue called accumulateQueue.
10. It pushes all the msgs into the accumulateQueue queue.
11. It measures the time it takes to accumulate all the messages in the accumulateQueue queue and prints the total time and operations per second for the accumulate operations.
12. It creates a new MessageQueue called pushPopThreadQueue.
13. It initializes a CompletableFuture called go.
14. It creates a thread called pusher that pushes all the msgs into the pushPopThreadQueue queue.
15. It creates a thread called popper that pops messages from the pushPopThreadQueue queue with a timeout of 10 milliseconds.
16. It measures the time it takes to perform the push and pop operations in separate threads and prints the total time and operations per second for these operations.
17. It initializes a new CompletableFuture called go2.
18. It creates a new MessageQueue called pushPopNowThreadQueue.
19. It creates a new thread called pusher that pushes all the msgs into the pushPopNowThreadQueue queue.
20. It creates a new thread called popper that pops messages from the pushPopNowThreadQueue queue without any timeout.
21. It measures the time it takes to perform the push and popnow operations in separate threads and prints the total time and operations per second for these operations.
22. It initializes a new CompletableFuture called go3.
23. It creates a new MessageQueue called pushAccumulateThreadQueue.
24. It creates a new thread called pusher that pushes all the msgs into the pushAccumulateThreadQueue queue.
25. It creates a new thread called popper that accumulates messages from the pushAccumulateThreadQueue queue until all messages have been accumulated.
26. It measures the time it takes to perform the push and accumulate operations in separate threads and prints the total time and operations per second for these operations.

The main method in the class io.nats.client.impl.MessageQueueBenchmark is used to benchmark the performance of various operations on a message queue.

The method starts by initializing an array of NatsMessage objects with a specified count of messages. It then runs a series of tests, measuring the time it takes to perform different operations on the message queue.

The method first warms up the message queue by pushing messages onto it. It then measures the time taken to perform a series of push operations on the queue. Next, it measures the time taken to perform a series of pop-now operations, which tries to pop a message from the queue with a timeout.

After that, the method initializes a new message queue and pushes messages onto it. It then measures the time taken to perform a series of accumulate operations, which continuously accumulates messages from the queue until it becomes empty.

The method then runs two threads concurrently - one pushing messages onto the queue and the other popping messages with a timeout. It measures the time taken for this scenario.

Finally, the method runs two threads - one pushing messages onto the queue and the other popping messages immediately. It measures the time taken for this scenario as well.

For each test scenario, the method outputs the total time taken for the operations, the time taken per operation, and the operations per second.

Overall, this method provides a benchmark for measuring the performance of various operations on a message queue implementation.

MessageProtocolCreationBenchmark

The MessageProtocolCreationBenchmark class is a public class used for benchmarking the creation of message protocols.

public static void main(String[] args) throws InterruptedException {
    int warmup = 1_000_000;
    int msgCount = 50_000_000;
    System.out.printf("### Running benchmarks with %s messages.\n", NumberFormat.getInstance().format(msgCount));
    for (int j = 0; j < warmup; j++) {
        new NatsMessage("subject", "replyTo", EMPTY_BODY);
    }
    long start = System.nanoTime();
    for (int j = 0; j < msgCount; j++) {
        new NatsMessage("subject", "replyTo", EMPTY_BODY);
    }
    long end = System.nanoTime();
    System.out.printf("\n### Total time to create %s non-utf8 messages for sending was %s ms\n\t%f ns/op\n\t%s op/sec\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), ((double) (end - start)) / ((double) (msgCount)), NumberFormat.getInstance().format(((double) (1_000_000_000L * msgCount)) / ((double) (end - start))));
    start = System.nanoTime();
    for (int j = 0; j < msgCount; j++) {
        new NatsMessage("subject", "replyTo", EMPTY_BODY);
    }
    end = System.nanoTime();
    System.out.printf("\n### Total time to create %s utf8 messages for sending was %s ms\n\t%f ns/op\n\t%s op/sec\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), ((double) (end - start)) / ((double) (msgCount)), NumberFormat.getInstance().format(((double) (1_000_000_000L * msgCount)) / ((double) (end - start))));
    start = System.nanoTime();
    for (int j = 0; j < msgCount; j++) {
        new ProtocolMessage(EMPTY_BODY);
    }
    end = System.nanoTime();
    System.out.printf("\n### Total time to create %s a protocol message was %s ms\n\t%f ns/op\n\t%s op/sec\n", NumberFormat.getInstance().format(msgCount), NumberFormat.getInstance().format((end - start) / 1_000_000L), ((double) (end - start)) / ((double) (msgCount)), NumberFormat.getInstance().format(((double) (1_000_000_000L * msgCount)) / ((double) (end - start))));
}

The main method in the class io.nats.client.impl.MessageProtocolCreationBenchmark performs the following steps:

1. Define the variables warmup and msgCount. warmup is set to 1,000,000 and msgCount is set to 50,000,000 (representing the number of messages to be processed).
2. Print a message indicating the number of messages that will be used for the benchmark.
3. Perform a warm-up phase by creating a new NatsMessage object 1,000,000 times using the values "subject", "replyTo", and EMPTY_BODY. This is done to warm-up the JVM and optimize any code before running the actual benchmark.
4. Record the start time using System.nanoTime().
5. Create a new NatsMessage object msgCount times using the same values as in the warm-up phase. This measures the time taken to create msgCount non-utf8 messages for sending.
6. Record the end time using System.nanoTime().
7. Print the total time taken to create msgCount non-utf8 messages in milliseconds, the average time taken per operation in nanoseconds, and the number of operations per second.
8. Repeat steps 5-7 with utf8 messages instead of non-utf8 messages.
9. Repeat steps 5-7 with a ProtocolMessage object instead of NatsMessage to measure the time taken to create a protocol message.
10. The benchmark is completed.

The main method in the MessageProtocolCreationBenchmark class is a benchmarking method. It measures the time taken to create different types of messages for sending.

Here's an overview of what the method does:

1. Sets the value for warmup and msgCount variables, which specify the number of warmup iterations and the total number of messages, respectively.
2. Prints the number of messages being benchmarked.
3. Performs warmup iterations, where each iteration creates a new NatsMessage object with specific parameters.
4. Starts timing.
5. Performs msgCount iterations, where each iteration creates a new NatsMessage object with specific parameters.
6. Stops timing and prints the total time taken to create the non-utf8 messages, the average time per operation, and the number of operations per second.
7. Resets timing and repeats steps 4-6 for creating utf8 messages.
8. Resets timing and repeats steps 4-6 for creating a protocol message.

Overall, this method helps measure the performance of creating different types of messages used in communication. The time taken for message creation, the average time per operation, and the number of operations per second are printed as benchmark results for each type of message.

NatsPackageScopeWorkarounds

The NatsPackageScopeWorkarounds class is a public class that provides workarounds for package scope limitations in the Nats package.

ParseTesting

ParseTesting is a public class that is used for testing parsing functionality. It contains various methods and properties to facilitate parsing operations.

public static void main(String[] args) throws InterruptedException {
    int iterations = 1_000_000;
    System.out.println("###");
    System.out.printf("### Running parse tests with %s msgs.\n", NumberFormat.getInstance().format(iterations));
    System.out.println("###");
    runTest(iterations, "+OK");
    runTest(iterations, "PONG");
    runTest(iterations, "INFO " + infoJson);
    runTest(iterations, "MSG longer.subject.abitlikeaninbox 22 longer.replyto.abitlikeaninbox 234");
    runTest(iterations, "-ERR some error with spaces in it");
}

The main method in the ParseTesting class performs the following steps:

1. Declare and initialize a variable iterations with the value 1_000_000.
2. Print the header information using System.out.println.
3. Use NumberFormat.getInstance().format(iterations) to format the value of iterations.
4. Print a message indicating the number of iterations being tested.
5. Call the runTest method with the specified iterations and the string "+OK".
6. Call the runTest method with the specified iterations and the string "PONG".
7. Call the runTest method with the specified iterations and the string "INFO " + infoJson, where infoJson is a variable that holds a JSON object.
8. Call the runTest method with the specified iterations and the string "MSG longer.subject.abitlikeaninbox 22 longer.replyto.abitlikeaninbox 234".
9. Call the runTest method with the specified iterations and the string "-ERR some error with spaces in it".

The runTest method is not defined in the given code snippet, so its functionality cannot be determined.

The main method in the ParseTesting class is used to run parse tests.

It first sets the number of iterations to 1 million and then prints out a header indicating the start of the parse tests, along with the number of messages involved in the test.

The runTest method is then called multiple times, passing different test strings as arguments. These test strings include various patterns such as "+OK", "PONG", "INFO" followed by infoJson, a longer subject and replyto message, and a message with an error.

Overall, this method executes parse tests with different message strings to test the parsing functionality of the software.

public static String[] splitCharBuffer(CharBuffer buffer) {
    ArrayList<String> list = new ArrayList<>();
    StringBuilder builder = new StringBuilder();
    while (buffer.hasRemaining()) {
        char c = buffer.get();
        if (c == ' ') {
            list.add(builder.toString());
            builder = new StringBuilder();
        } else {
            builder.append(c);
        }
    }
    if (builder.length() > 0) {
        list.add(builder.toString());
    }
    return list.toArray(new String[0]);
}

The splitCharBuffer method in the ParseTesting class, defined in the io.nats.client.impl package, is used to split a CharBuffer into an array of strings. Here is a step-by-step description of what the method does:

1. Create an empty ArrayList to store the individual strings extracted from the CharBuffer.
2. Create an empty StringBuilder as a temporary container for each individual string.
3. Start a loop that continues as long as there are remaining characters in the CharBuffer.
4. Retrieve the next character from the CharBuffer using the get() method, and assign it to the variable c.
5. Check if the character c is a space (' ').
6. If the character c is a space, convert the contents of the StringBuilder to a string using the toString() method, and add it to the ArrayList.
7. Create a new empty StringBuilder to start collecting a new string.
8. If the character c is not a space, append the character to the StringBuilder using the append() method.
9. Repeat steps 4 to 8 until all characters in the CharBuffer have been processed.
10. Check if the StringBuilder still contains characters that have not been added to the ArrayList.
11. If so, convert the remaining characters in the StringBuilder to a string using the toString() method, and add it to the ArrayList.
12. Convert the ArrayList to an array of strings using the toArray(T[] a) method with an empty array as the argument.
13. Return the resulting array of strings.

This method essentially splits the characters in the CharBuffer by spaces, creating an array of strings where each element represents a word or a group of non-space characters.

The method splitCharBuffer in class ParseTesting is used to split a CharBuffer into an array of strings based on spaces (' ') as the delimiter.

Here is a brief description of what the method does:

1. Creates an empty ArrayList to hold the individual strings.
2. Initializes a StringBuilder to build each string.
3. Iterates through each character in the CharBuffer using a while loop.
4. Checks if the current character is a space (' ').
   • If it is a space, the contents of the StringBuilder are added to the ArrayList as a string, and the StringBuilder is cleared to build a new string.
   • If it is not a space, the character is appended to the StringBuilder.
5. After the loop, the contents of the StringBuilder are added to the ArrayList if it is not empty.
6. Converts the ArrayList into an array of strings using the toArray method.
7. Returns the array of strings.

public static String grabNextWithBuilder(CharBuffer buffer) {
    StringBuilder builder = new StringBuilder();
    while (buffer.hasRemaining()) {
        char c = buffer.get();
        if (c == ' ') {
            return builder.toString();
        } else {
            builder.append(c);
        }
    }
    return builder.toString();
}

The method grabNextWithBuilder defined in the class io.nats.client.impl.ParseTesting takes a CharBuffer named buffer as input and returns a String.

Here is a step-by-step description of what the method does:

1. Create a new StringBuilder called builder to store characters from the buffer.

2. Enter a loop that continues as long as the buffer has remaining characters.

3. Get the next character from the buffer and assign it to the variable c.

4. Check if the character c is equal to a space (' '). If it is, then it means the method has encountered a space and it should stop reading characters from the buffer. In this case, the method returns the current contents of the builder as a String.

5. If the character c is not equal to a space, then it means the method has not encountered a space yet and it should continue reading characters from the buffer. In this case, the character c is appended to the builder.

6. After appending the character c to the builder, the loop continues to the next iteration to get the next character from the buffer.

7. If the loop completes without encountering a space character, it means that the entire buffer has been consumed. In this case, the method returns the current contents of the builder as a String.

This method essentially reads characters from the buffer until it encounters a space character or reaches the end of the buffer. It then returns the read characters as a String.

The grabNextWithBuilder method defined in the ParseTesting class of the io.nats.client.impl package is a static method that takes a CharBuffer as its parameter.

The purpose of this method is to extract the next string from the given CharBuffer. It does this by iterating over each character in the buffer. If a space character (' ') is encountered, the method returns the accumulated characters in the StringBuilder as a string. If any other character is encountered, it is appended to the StringBuilder.

After processing all characters in the buffer, the method returns the accumulated characters in the StringBuilder as a string, regardless of whether a space was encountered or not.

public static String grabNextWithSubsequence(CharBuffer buffer) {
    if (!buffer.hasRemaining()) {
        return null;
    }
    int start = buffer.position();
    while (buffer.hasRemaining()) {
        char c = buffer.get();
        if (c == ' ') {
            int end = buffer.position();
            buffer.position(start);
            //don't grab the space
            CharBuffer slice = buffer.subSequence(0, end - start - 1);
            buffer.position(end);
            return slice.toString();
        }
    }
    buffer.position(start);
    String retVal = buffer.toString();
    buffer.position(buffer.limit());
    return retVal;
}

The grabNextWithSubsequence method in the ParseTesting class is used to extract a substring from a given CharBuffer until the next occurrence of a space character (' '). Here is a step-by-step description of what the method does:

1. Check if the input buffer has any remaining characters. If it doesn't, return null.

2. Get the current position of the buffer and assign it to the start variable. This is the starting position from where the substring will be extracted.

3. Enter a while loop that runs until there are no more characters remaining in the buffer.

4. Get the next character in the buffer and assign it to the variable c.

5. Check if the character c is equal to a space (' '). If it is, it means we have found the next occurrence of a space character.

6. Get the current position of the buffer and assign it to the end variable. This is the ending position of the substring.

7. Set the position of the buffer back to the start position so that we can create a subsequence from the original buffer starting at the start position.

8. Create a new CharBuffer named slice using the subSequence method, which extracts a subsequence of characters from the buffer starting at the start position and ending at end - start - 1 position. This is done to exclude the space character.

9. Set the position of the buffer to the end position so that we can continue iterating through the remaining characters.

10. Convert the slice CharBuffer to a String using the toString method and assign it to the retVal variable.

11. Set the position of the buffer to its limit, which effectively marks the end of the buffer.

12. Return the retVal string, which is the extracted substring.

By following these steps, the grabNextWithSubsequence method efficiently extracts the next substring from the CharBuffer until the next space character is encountered.

The grabNextWithSubsequence method in the io.nats.client.impl.ParseTesting class is used to extract the next subsequence of characters from a given CharBuffer object.

Here's how it works:

• It starts by checking if the given CharBuffer has any remaining characters. If not, it returns null.
• If there are remaining characters, it captures the starting position of the buffer.
• It then iterates over the buffer, character by character, until it encounters a space character.
• Once a space is found, it captures the ending position of the subsequence and creates a new CharBuffer slice from the original buffer, excluding the space character.
• Finally, it resets the position of the original buffer to the starting position, returns the string representation of the subsequence, and sets the position of the buffer to its limit.

In summary, this method extracts the next subsequence of characters from a CharBuffer until it reaches a space character, excluding the space itself.

public static CharSequence grabNextAsSubsequence(CharBuffer buffer) {
    if (!buffer.hasRemaining()) {
        return null;
    }
    int start = buffer.position();
    while (buffer.hasRemaining()) {
        char c = buffer.get();
        if (c == ' ') {
            int end = buffer.position();
            buffer.position(start);
            //don't grab the space
            CharBuffer slice = buffer.subSequence(0, end - start - 1);
            buffer.position(end);
            return slice;
        }
    }
    buffer.position(start);
    CharSequence retVal = buffer.subSequence(0, buffer.remaining());
    buffer.position(buffer.limit());
    return retVal;
}

Description: This method is used to grab the next character sequence from the input CharBuffer as a subsequence. It stops when it encounters a space character (' ') and returns the subsequence excluding the space. If the buffer is empty or contains only spaces, it returns null.

Parameters:

• buffer: The input CharBuffer from which to grab the next subsequence.

Returns:

• CharSequence : The grabbed subsequence excluding space, or null if the buffer is empty or contains only spaces.

Steps: The method performs the following steps:

1. Check if the buffer is empty (buffer.hasRemaining()). If not, proceed to the next step. Otherwise, return null.

2. Get the current position of the buffer (start = buffer.position()).

3. Enter a while loop, iterating until the buffer has no more remaining characters.

4. Get the next character from the buffer (c = buffer.get()).

5. Check if the character is a space (' '). If true, go to the next step. Otherwise, continue with the next iteration of the loop.

6. Get the current position of the buffer (end = buffer.position()).

7. Set the position of the buffer back to the start position (buffer.position(start)).

8. Create a new CharBuffer slice from the original buffer, starting from index 0 and ending at end - start - 1 (excluding the space). Assign it to slice variable.

9. Set the position of the buffer to the end position (buffer.position(end)).

10. Return the grabbed subsequence (slice).

11. If the loop ends without encountering a space character, set the buffer position back to the start position (buffer.position(start)).

12. Create a new CharBuffer subsequence from the original buffer, starting from index 0 and ending at the remaining characters in the buffer. Assign it to retVal variable.

13. Set the buffer position to the buffer limit (buffer.position(buffer.limit())).

14. Return the grabbed subsequence (retVal).

The grabNextAsSubsequence method in the ParseTesting class takes a CharBuffer as input and returns the next sequence of characters in the buffer until it encounters a space character (' ').

It starts by checking if the buffer has remaining characters. If it doesn't, it returns null. Otherwise, it records the starting position of the buffer.

The method then iterates over the remaining characters in the buffer. For each character, it checks if it is a space character. If it is, it records the ending position of the buffer and creates a subsequence (excluding the space character) from the starting position to the ending position. The buffer's position is then updated to the ending position, and the subsequence is returned.

If no space character is found, the method resets the buffer's position to the starting position, and creates a subsequence from the starting position to the end of the buffer. The buffer's position is then set to its limit, and the subsequence is returned.

In summary, this method retrieves the next sequence of characters from a CharBuffer until it encounters a space character, excluding the space character from the resulting subsequence.

public static String grabNextWithCharArray(CharBuffer buffer) {
    int remaining = buffer.remaining();
    if (remaining == 0) {
        return null;
    }
    int i = 0;
    while (remaining > 0) {
        char c = buffer.get();
        if (c == ' ') {
            return new String(buff, 0, i);
        } else {
            buff[i] = c;
            i++;
        }
        remaining--;
    }
    return new String(buff, 0, i);
}

The grabNextWithCharArray method in the ParseTesting class is a public static method that takes a CharBuffer as its parameter. The purpose of this method is to extract the next string from the CharBuffer.

• buffer: A CharBuffer object that contains the characters to be processed.

• If the buffer is empty (i.e., remaining == 0), the method returns null.
• If the next character in the buffer is a space character, the method returns a new String object containing the characters from buff[0] to buff[i-1] (where buff is an array of characters).
• Otherwise, the method continues to read characters from the buffer until a space character is encountered or the end of the buffer is reached. It then returns a new String object containing the characters from buff[0] to buff[i-1].

1. The method starts by retrieving the remaining number of characters in the buffer using the remaining() method of the CharBuffer.
2. If the remaining count is equal to 0, the method returns null, indicating that there are no more characters to process.
3. If there are remaining characters, the method initializes a counter variable i as 0 to keep track of the current position in the buff array (initialized elsewhere in the code and not shown).
4. The method enters a loop that will continue until there are no remaining characters in the buffer.
5. In each iteration of the loop, the method retrieves the next character from the buffer using the get() method of the CharBuffer and assigns it to the variable c.
6. If the character c is a space character, the method creates a new String object using the characters stored in the buff array from index 0 to i-1 (these characters represent the parsed string), and returns it.
7. If the character c is not a space, it is stored in the buff array at the current position i, and the i counter is incremented by 1.
8. The method then decrements the remaining count by 1.
9. The loop continues until the remaining count reaches 0, indicating that all characters from the buffer have been processed.
10. If no space character was encountered in the buffer, the method creates a new String object using the characters stored in the buff array from index 0 to i-1 and returns it.

Note: The code provided does not show the declaration and initialization of the buff array, which is assumed to be present in the class and accessible to this method.

The method grabNextWithCharArray is a public static method defined in the class io.nats.client.impl.ParseTesting. It takes a CharBuffer object as a parameter.

The purpose of this method is to extract the next word delimited by a space character (' ') from the CharBuffer and return it as a String. If there are no more characters left in the buffer, it returns null.

Internally, the method iterates over the characters in the CharBuffer using a while loop. It reads each character one by one and checks if it is a space character. If it is, it creates a new String object from the characters read so far using the new String(buff, 0, i) constructor and returns it.

If the character is not a space, it adds the character to an internal character array buff at the current position i. After each character is read, remaining is decremented and i is incremented. This process continues until either a space character is encountered or there are no more characters left in the buffer.

Finally, if the end of the buffer is reached without encountering a space character, it creates a new String object from the characters read so far using new String(buff, 0, i) and returns it.

Note: The buff array is not declared in the given code snippet, so the method assumes that it is already declared and accessible within the class scope.

public static String protocolFor(char[] chars, int length) {
    if (length == 3) {
        if (chars[0] == '+' && chars[1] == 'O' && chars[2] == 'K') {
            return OP_OK;
        } else if (chars[0] == 'M' && chars[1] == 'S' && chars[2] == 'G') {
            return OP_MSG;
        } else {
            return null;
        }
    } else if (length == 4) {
        // order for uniqueness
        if (chars[1] == 'I' && chars[0] == 'P' && chars[2] == 'N' && chars[3] == 'G') {
            return OP_PING;
        } else if (chars[0] == 'P' && chars[1] == 'O' && chars[2] == 'N' && chars[3] == 'G') {
            return OP_PONG;
        } else if (chars[0] == '-' && chars[1] == 'E' && chars[2] == 'R' && chars[3] == 'R') {
            return OP_ERR;
        } else if (chars[2] == 'F' && chars[0] == 'I' && chars[1] == 'N' && chars[3] == 'O') {
            return OP_INFO;
        } else {
            return null;
        }
    } else {
        return null;
    }
}

The protocolFor method in the io.nats.client.impl.ParseTesting class is responsible for determining the protocol type based on the input chars and length.

The method accepts two parameters - chars which is a character array and length which denotes the size of the array.

If the length of the array is 3, it checks if the first character is '+', the second character is 'O', and the third character is 'K'. If it matches, it returns the value OP_OK. Else, it checks if the first character is 'M', the second character is 'S', and the third character is 'G'. If it matches, it returns the value OP_MSG. If none of the conditions are met, it returns null.

If the length of the array is 4, it checks for the following conditions:

• If the second character is 'I', the first character is 'P', the third character is 'N' and the fourth character is 'G', it returns the value OP_PING.
• If the first character is 'P', the second character is 'O', the third character is 'N' and the fourth character is 'G', it returns the value OP_PONG.
• If the first character is '-', the second character is 'E', the third character is 'R', and the fourth character is 'R', it returns the value OP_ERR.
• If the third character is 'F', the first character is 'I', the second character is 'N', and the fourth character is 'O', it returns the value OP_INFO. If none of the conditions are met, it returns null.

If the length of the array is neither 3 nor 4, it returns null.

Overall, this method determines the protocol type based on the characters in the chars array and returns the corresponding protocol value.

The method protocolFor in the class ParseTesting is used to determine the protocol for the given array of characters (chars) and its length (length). It checks the length of the array and performs different checks based on the length to determine the protocol.

If the length is 3, it checks if the characters at indexes 0, 1, and 2 are equal to '+', 'O', and 'K' respectively. If they are, it returns the constant OP_OK. Otherwise, it checks if the characters represent the 'MSG' protocol and returns OP_MSG. If none of these conditions are met, it returns null.

If the length is 4, it performs similar checks for different protocols. For example, it checks if the characters at indexes 0, 1, 2, and 3 are equal to 'P', 'I', 'N', and 'G' respectively to determine the 'PING' protocol. It checks for other protocols like 'PONG', 'ERR', and 'INFO' in a similar manner. If none of these conditions are met, it returns null.

If the length is neither 3 nor 4, it returns null.

In summary, the method protocolFor is used to determine the protocol based on the given array of characters and its length. It returns the corresponding protocol constant or null if the protocol cannot be determined.

public static String grabProtocol(CharBuffer buffer) {
    int remaining = buffer.remaining();
    if (remaining == 0) {
        return null;
    }
    int i = 0;
    while (remaining > 0) {
        char c = buffer.get();
        if (c == ' ') {
            return protocolFor(buff, i);
        } else {
            buff[i] = c;
            i++;
        }
        remaining--;
    }
    return protocolFor(buff, i);
}

The grabProtocol method takes a CharBuffer as a parameter and returns a string representing the protocol.

1. Check the number of remaining characters in the buffer using buffer.remaining(). If there are no remaining characters (i.e., remaining == 0), return null.

2. Initialize a variable i with a value of 0. This variable will be used as an index to store characters in a temporary buffer.

3. Enter a loop while there are remaining characters in the buffer (i.e., remaining > 0).

4. Get the next character from the buffer using buffer.get() and assign it to a variable c.

5. Check if the character c is a space character.

   a. If it is a space character, call a method protocolFor with the temporary buffer buff (not shown in the provided code) and the index i as arguments, and return the result of this method call as the protocol string.

   b. If it is not a space character, store the character c in the temporary buffer buff at index i.

6. Decrement the remaining variable by 1, indicating that one character has been processed.

7. After exiting the loop, call the protocolFor method again with the temporary buffer buff and the index i as arguments, and return the result as the protocol string.

Here is a modified version of the provided code with the missing buff variable declaration and the protocolFor method called with the correct arguments:

public static String grabProtocol(CharBuffer buffer) {
    int remaining = buffer.remaining();
    if (remaining == 0) {
        return null;
    }

    char[] buff = new char[remaining]; // Temporary buffer to store characters
    int i = 0;

    while (remaining > 0) {
        char c = buffer.get();

        if (c == ' ') {
            return protocolFor(buff, i);
        } else {
            buff[i] = c;
            i++;
        }
        remaining--;
    }

    return protocolFor(buff, i);
}

The grabProtocol method in the ParseTesting class, defined in the io.nats.client.impl package, takes a CharBuffer as input.

It first checks the number of remaining characters in the buffer. If it is zero, it returns null.

Otherwise, it initializes a counter variable i and enters a loop while there are remaining characters in the buffer. Inside the loop, it retrieves each character from the buffer and checks if it is a space (' '). If it is, it invokes the protocolFor method with the buff array and the current value of i as arguments, and returns the result.

If the current character is not a space, it stores it in the buff array at index i and increments i by 1.

Regardless of whether a space character is found or not, the method decrements the remaining variable by 1 after processing each character.

Finally, if the loop completes without encountering a space character, the method again invokes the protocolFor method with the buff array and the final value of i as arguments, and returns the result.

This method is expected to determine the protocol from the input CharBuffer by extracting the string until the first space (' ') character is encountered.

public static void runTest(int iterations, String serverMessage) {
    Pattern space = Pattern.compile(" ");
    ByteBuffer protocolBuffer = ByteBuffer.allocate(32 * 1024);
    protocolBuffer.put(serverMessage.getBytes(StandardCharsets.UTF_8));
    protocolBuffer.flip();
    CharBuffer buffer = StandardCharsets.UTF_8.decode(protocolBuffer);
    String pl = buffer.toString();
    buffer.rewind();
    protocolBuffer.rewind();
    String[] newversion = splitCharBuffer(buffer);
    String[] oldversion = space.split(pl);
    buffer.rewind();
    ArrayList<String> opAware = new ArrayList<>();
    CharSequence s = null;
    while ((s = grabNext(buffer)) != null) {
        opAware.add(s.toString());
    }
    String[] opAwareArray = opAware.toArray(new String[0]);
    System.out.printf("### Parsing server string: %s\n", serverMessage);
    boolean newOk = Arrays.equals(newversion, oldversion);
    System.out.println("### Old and new versions are equal: " + newOk);
    if (!newOk) {
        System.exit(-1);
    }
    boolean protoOk = Arrays.equals(opAwareArray, oldversion);
    System.out.println("### Old and op-aware versions are equal: " + protoOk);
    if (!protoOk) {
        System.exit(-1);
    }
    long start = System.nanoTime();
    for (int i = 0; i < iterations; i++) {
        StandardCharsets.UTF_8.decode(protocolBuffer).toString();
        protocolBuffer.rewind();
    }
    long end = System.nanoTime();
    System.out.printf("### %s raw utf8 decode/sec.\n", NumberFormat.getInstance().format(1_000_000_000L * iterations / (end - start)));
    start = System.nanoTime();
    for (int i = 0; i < iterations; i++) {
        String protocolLine = StandardCharsets.UTF_8.decode(protocolBuffer).toString();
        space.split(protocolLine);
        protocolBuffer.rewind();
    }
    end = System.nanoTime();
    System.out.printf("### %s old parses/sec.\n", NumberFormat.getInstance().format(1_000_000_000L * iterations / (end - start)));
    start = System.nanoTime();
    for (int i = 0; i < iterations; i++) {
        CharBuffer charBuffer = StandardCharsets.UTF_8.decode(protocolBuffer);
        splitCharBuffer(charBuffer);
        protocolBuffer.rewind();
    }
    end = System.nanoTime();
    System.out.printf("### %s new parses/sec.\n", NumberFormat.getInstance().format(1_000_000_000L * iterations / (end - start)));
    start = System.nanoTime();
    for (int i = 0; i < iterations; i++) {
        CharBuffer charBuffer = StandardCharsets.UTF_8.decode(protocolBuffer);
        String op = grabProtocol(charBuffer);
        switch(op) {
            case OP_MSG:
                //subject
                grabNext(charBuffer);
                // sid
                grabNext(charBuffer);
                // replyto or length
                grabNext(charBuffer);
                // length or null
                grabNext(charBuffer);
                break;
            case OP_ERR:
                grabTheRest(charBuffer);
                break;
            case OP_OK:
            case OP_PING:
            case OP_PONG:
            case OP_INFO:
                grabTheRest(charBuffer);
                break;
            default:
                break;
        }
        protocolBuffer.rewind();
    }
    end = System.nanoTime();
    System.out.printf("### %s op-aware parses/sec.\n", NumberFormat.getInstance().format(1_000_000_000L * iterations / (end - start)));
    System.out.println();
}

The runTest method in the io.nats.client.impl.ParseTesting class is performing the following steps:

1. It receives two parameters: iterations (the number of iterations to run the test) and serverMessage (the message to be parsed).
2. It creates a Pattern object called space to match a space character.
3. It creates a ByteBuffer called protocolBuffer with a size of 32 * 1024 bytes.
4. It puts the bytes of the serverMessage string into the protocolBuffer using the UTF-8 encoding.
5. It flips the protocolBuffer to prepare for reading.
6. It decodes the bytes in the protocolBuffer using the UTF-8 charset to create a CharBuffer called buffer.
7. It converts the buffer to a string pl.
8. It rewinds the buffer and protocolBuffer to prepare for reading.
9. It splits the buffer into an array of strings called newversion using a custom method splitCharBuffer.
10. It splits the pl string into an array of strings called oldversion by matching spaces.
11. It rewinds the buffer and creates an ArrayList called opAware.
12. It enters a loop where it grabs the next sequence of characters from the buffer and adds it to the opAware list until there are no more sequences.
13. It converts the opAware list to an array of strings called opAwareArray.
14. It prints the serverMessage in a formatted string.
15. It checks if newversion and oldversion arrays are equal and assigns the result to newOk.
16. It prints whether the old and new versions are equal.
17. If the old and new versions are not equal, it exits the program with a status of -1.
18. It checks if opAwareArray and oldversion arrays are equal and assigns the result to protoOk.
19. It prints whether the old and op-aware versions are equal.
20. If the old and op-aware versions are not equal, it exits the program with a status of -1.
21. It initializes a variable start with the current system time in nanoseconds.
22. It enters a loop where it decodes the protocolBuffer using UTF-8 encoding and converts it to a string, and then rewinds the protocolBuffer, for each iteration.
23. It calculates the elapsed time in nanoseconds by subtracting start from the current system time and assigns it to end.
24. It prints the number of raw UTF-8 decodes per second.
25. It initializes start with the current system time in nanoseconds.
26. It enters a loop where it decodes the protocolBuffer, converts it to a string, splits the resulting string using spaces, and then rewinds the protocolBuffer, for each iteration.
27. It calculates the elapsed time in nanoseconds by subtracting start from the current system time and assigns it to end.
28. It prints the number of old parses per second.
29. It initializes start with the current system time in nanoseconds.
30. It enters a loop where it decodes the protocolBuffer and splitCharBuffer the resulting CharBuffer, and then rewinds the protocolBuffer, for each iteration.
31. It calculates the elapsed time in nanoseconds by subtracting start from the current system time and assigns it to end.
32. It prints the number of new parses per second.
33. It initializes start with the current system time in nanoseconds.
34. It enters a loop where it decodes the protocolBuffer, gets the next operation from the resulting CharBuffer, and performs different actions based on the operation, and then rewinds the protocolBuffer, for each iteration.
35. It calculates the elapsed time in nanoseconds by subtracting start from the current system time and assigns it to end.
36. It prints the number of op-aware parses per second.
37. It prints an empty line.

The runTest method in the ParseTesting class is a performance test method used to compare different parsing approaches for a given server message.

This method takes in two parameters: iterations (representing the number of iterations to run the test) and serverMessage (representing the message received from the server).

The method starts by initializing the necessary variables, such as space (to split strings by space), protocolBuffer (a byte buffer to store the server message), and buffer (a character buffer decoded from the byte buffer using UTF-8). It also splits the buffer and pl (the decoded server message) into arrays of strings.

Next, it creates an ArrayList called opAware and iterates over the characters in the buffer, adding them to the opAware list as separate strings. It then converts the opAware list to an array.

The method then prints some debug information about the server message and compares the old and new versions of the parsed message. If they are not equal, the method exits with an error code.

It then performs several benchmarking tests on the parsing approaches. It measures the time it takes to decode the server message multiple times using the old and new parsing methods, as well as an "op-aware" parsing method. It also measures the time it takes to grab specific protocol elements based on their operation type.

Finally, it prints the results of the benchmarking tests.

The purpose of this method is to assess the performance of different parsing approaches and determine the most efficient one for decoding and processing server messages.

StringListReader

The StringListReader class is an abstract class that extends the AbstractListReader. It provides functionality to read a list of strings.

void processItems(List items)

@Override
void processItems(List<JsonValue> items) {
    for (JsonValue v : items) {
        if (v.string != null) {
            strings.add(v.string);
        }
    }
}

This method is implemented in the io.nats.client.impl.StringListReader class and overrides the processItems method from its parent class.

The purpose of this method is to process a list of JsonValue items and add any strings found in the items to the strings list.

1. Iterate over each JsonValue in the items list
2. For each JsonValue, check if the string field is not null
3. If the string field is not null, add it to the strings list

The processItems method, defined in the StringListReader class in the io.nats.client.impl package, is an overridden method that takes a List of JsonValues as input and performs the following operations:

• It iterates over each JsonValue v in the provided list.
• It checks if the string property of v is not null.
• If the string property is not null, it adds the value of v.string to the strings collection.

In summary, the processItems method collects non-null string values from a list of JsonValues and adds them to the strings collection.

ListRequestEngine

The ListRequestEngine class is an extension of the ApiResponse class. It represents a request engine for listing items and provides functionalities for managing and retrieving lists of items.

byte[] internalNextJson(String fieldName, String filter) {
    if (hasMore()) {
        if (filter == null) {
            return noFilterJson();
        }
        return (OFFSET_JSON_START + nextOffset() + ",\"" + fieldName + "\":\"" + filter + "\"}").getBytes(StandardCharsets.US_ASCII);
    }
    return null;
}
```## NatsConsumer

**NatsConsumer**

The `NatsConsumer` class is an abstract class that implements the `Consumer` interface. This class is used for consuming messages from a NATS messaging system. It provides the necessary functionality for connecting to a NATS server, subscribing to specific topics, and processing messages received from the server. Users can extend this class to create their own custom consumer implementations for handling NATS messages.
### boolean hasReachedPendingLimits() 
```java
boolean hasReachedPendingLimits() {
    long ml = maxMessages.get();
    if (ml > 0 && getPendingMessageCount() >= ml) {
        return true;
    }
    long bl = maxBytes.get();
    return bl > 0 && getPendingByteCount() >= bl;
}

The hasReachedPendingLimits() method, defined in the NatsConsumer class within the io.nats.client.impl package, is responsible for determining whether the consumer has reached its pending limits.

Here's a step-by-step breakdown of what the method does:

1. Retrieve the value of the maxMessages variable, which represents the maximum number of messages allowed to be pending. Store it in the ml variable.
2. Check if ml is greater than 0 and if the number of pending messages, obtained from the getPendingMessageCount() method, is greater than or equal to ml.
3. If the above condition is true, it means that the pending message count has reached or exceeded the maximum allowed limit. In this case, the method returns true, indicating that the pending limits have been reached.
4. If the above condition is false, retrieve the value of the maxBytes variable, which represents the maximum number of bytes allowed to be pending. Store it in the bl variable.
5. Check if bl is greater than 0 and if the number of pending bytes, obtained from the getPendingByteCount() method, is greater than or equal to bl.
6. If the above condition is true, it means that the pending byte count has reached or exceeded the maximum allowed limit. In this case, the method returns true, indicating that the pending limits have been reached.
7. If both conditions in step 3 and step 6 are false, it means that the pending limits have not been reached. In this case, the method returns false.

The hasReachedPendingLimits() method essentially checks if either the maximum number of pending messages or bytes has been reached and returns true, otherwise it returns false.

The hasReachedPendingLimits method in the NatsConsumer class is a boolean method that determines whether the consumer has reached its pending message or byte limits.

It first retrieves the maximum number of messages (ml) using the maxMessages field. If ml is greater than 0 and the number of pending messages is equal to or exceeds ml, the method returns true.

If the number of pending messages is below ml or ml is 0, the method proceeds to check the maximum number of bytes (bl) using the maxBytes field. If bl is greater than 0 and the number of pending bytes is equal to or exceeds bl, the method returns true.

If both the maximum number of messages and maximum number of bytes are below their respective limits, the method returns false.

public CompletableFuture<Boolean> drain(Duration timeout) throws InterruptedException {
    if (!this.isActive() || this.connection == null) {
        throw new IllegalStateException("Consumer is closed");
    }
    if (isDraining()) {
        return this.getDrainingFuture();
    }
    Instant start = Instant.now();
    final CompletableFuture<Boolean> tracker = new CompletableFuture<>();
    this.markDraining(tracker);
    this.sendUnsubForDrain();
    try {
        // Flush and wait up to the timeout
        this.connection.flush(timeout);
    } catch (TimeoutException e) {
        this.connection.processException(e);
    }
    this.markUnsubedForDrain();
    // Wait for the timeout or the pending count to go to 0, skipped if conn is
    // draining
    connection.getExecutor().submit(() -> {
        try {
            Instant now = Instant.now();
            while (timeout == null || timeout.equals(Duration.ZERO) || Duration.between(start, now).compareTo(timeout) < 0) {
                if (this.isDrained()) {
                    break;
                }
                // Sleep 1 milli
                Thread.sleep(1);
                now = Instant.now();
            }
            this.cleanUpAfterDrain();
        } catch (InterruptedException e) {
            this.connection.processException(e);
        } finally {
            tracker.complete(this.isDrained());
        }
    });
    return getDrainingFuture();
}

The drain method is defined in the class io.nats.client.impl.NatsConsumer and takes a Duration parameter called timeout. It returns a CompletableFuture<Boolean>.

Here is a step-by-step description of what the drain method does based on its body:

1. Check if the consumer is active and if the connection is not null. If either condition is false, throw an IllegalStateException with the message "Consumer is closed".
2. Check if the consumer is already draining. If it is, return the result of getDrainingFuture(), which is a CompletableFuture<Boolean>.
3. Get the current timestamp and create a new CompletableFuture<Boolean> called tracker.
4. Mark the consumer as draining using the markDraining method and pass in the tracker as an argument.
5. Send an unsubscribe message for draining using the sendUnsubForDrain method.
6. Try to flush the connection and wait up to the specified timeout. If a TimeoutException occurs, handle it by calling the processException method of the connection.
7. Mark the consumer as unsubscribed for draining using the markUnsubedForDrain method.
8. Submit a task to the connection's executor to run the following code:
   • Create a new timestamp called now.
   • Enter a loop that continues while the timeout is null, equal to Duration.ZERO, or the difference between start and now is less than timeout.
   • Inside the loop, check if the consumer is drained. If it is, break out of the loop.
   • Sleep for 1 millisecond.
   • Update the value of now to the current timestamp.
   • After the loop ends, call the cleanUpAfterDrain method.
   • Finally, complete the tracker with the boolean value indicating if the consumer is drained.
9. Return the result of getDrainingFuture().

Please note that this description is based on the provided code snippet, and further details about the class and its dependencies may be necessary to fully understand the functionality of the drain method.

The drain method in the NatsConsumer class is used to gracefully stop consuming messages from a NATS server. When called, it performs the following steps:

1. Checks if the consumer is active and if a connection to the NATS server is established. If not, it throws an IllegalStateException.
2. Checks if the consumer is already in the process of draining (i.e., stopping). If it is, it returns a future that represents the current draining state.
3. Marks the consumer as draining and creates a new CompletableFuture object to track its draining state.
4. Sends an unsubscribe message to the NATS server to stop receiving new messages.
5. Flushes the connection and waits for the specified timeout duration for pending messages to be sent.
6. If the flush times out, it processes the timeout exception in the connection.
7. Marks the consumer as unsubscribed for draining.
8. Spawns a new thread that continuously checks if the draining has completed within the specified timeout.
9. Within this thread, it checks if the draining has completed or if the timeout duration has passed. If so, it breaks out of the loop.
10. After the loop exits, it performs necessary cleanup tasks after the draining process.
11. Finally, it sets the draining future to the current draining state and returns it.

Overall, the drain method allows for controlled termination of message consumption from the NATS server, ensuring that any pending messages are sent before stopping and providing a way to track the draining state.

StringAuthHandler

The StringAuthHandler class is an implementation of the AuthHandler interface. It provides functionality for authenticating users based on a string representation of their credentials.

public byte[] sign(byte[] nonce) {
    try {
        NKey nkey = NKey.fromSeed(this.nkey);
        byte[] sig = nkey.sign(nonce);
        nkey.clear();
        return sig;
    } catch (Exception exp) {
        throw new IllegalStateException("problem signing nonce", exp);
    }
}

The sign method in the StringAuthHandler class in the io.nats.client.impl package is responsible for generating a signature for a given nonce.

Here is a step-by-step description of what the method is doing:

1. Convert the nkey string (presumably containing a seed) into an NKey object using the fromSeed method.
2. Generate a signature for the given nonce by calling the sign method on the NKey object. The generated signature will be stored in a byte array.
3. Clear the NKey object by calling the clear method. This is done to securely dispose of any sensitive information stored in the NKey object.
4. Return the generated signature as a byte array.

If any exceptions occur during the process, an IllegalStateException is thrown with the message "problem signing nonce" and the corresponding exception as the cause.

The method sign in the StringAuthHandler class is used to sign a given nonce using an NKey.

To do this, it first creates an instance of the NKey class using the nkey property of the StringAuthHandler object. Then, it calls the sign method of the NKey instance, passing the nonce as a parameter. The returned value is the signature of the nonce.

After the signature is obtained, the clear method is called on the NKey instance to clear any sensitive information related to the key.

Finally, the method returns the signature as a byte array.

If any exception occurs during the signing process, an IllegalStateException is thrown with an error message indicating the problem encountered.

NatsJetStreamImpl

The NatsJetStreamImpl class is an implementation of the NatsJetStreamConstants interface. It provides functionality related to the NATS JetStream feature.

// Management that is also needed by regular context // ---------------------------------------------------------------------------------------------------- ConsumerInfo _getConsumerInfo(String streamName, String consumerName) throws IOException, JetStreamApiException

// ----------------------------------------------------------------------------------------------------
// Management that is also needed by regular context
// ----------------------------------------------------------------------------------------------------
ConsumerInfo _getConsumerInfo(String streamName, String consumerName) throws IOException, JetStreamApiException {
    String subj = String.format(JSAPI_CONSUMER_INFO, streamName, consumerName);
    Message resp = makeRequestResponseRequired(subj, null, jso.getRequestTimeout());
    return new ConsumerInfo(resp).throwOnHasError();
}

The _getConsumerInfo method is defined in the NatsJetStreamImpl class in the io.nats.client.impl package. It takes two parameters: streamName and consumerName.

Below is a step-by-step description of what this method does based on its body:

1. It creates a subject by formatting the streamName and consumerName using the JSAPI_CONSUMER_INFO format string. This subject is used to request information about a specific consumer in a specific JetStream stream.

String subj = String.format(JSAPI_CONSUMER_INFO, streamName, consumerName);

2. It makes a request to the NATS server and expects a response. The makeRequestResponseRequired method, which is not shown in the code snippet, is responsible for sending the request and handling the response. The subj parameter is used as the subject for the request, and the null parameter indicates that the request does not include any additional payload data. The jso.getRequestTimeout() method is used to determine the timeout for the request.

Message resp = makeRequestResponseRequired(subj, null, jso.getRequestTimeout());

3. It creates a ConsumerInfo object using the response message obtained from the previous step. The ConsumerInfo class is responsible for parsing and representing the consumer information received from the JetStream server.

return new ConsumerInfo(resp).throwOnHasError();

4. It returns the ConsumerInfo object. Before returning, it checks if the ConsumerInfo object contains any error. If an error is detected, it throws a JetStreamApiException with details about the error.

That's the step-by-step description of the _getConsumerInfo method based on its body.

The _getConsumerInfo method in the class io.nats.client.impl.NatsJetStreamImpl is responsible for retrieving the information of a specific consumer within a given stream.

Given the streamName and consumerName as parameters, the method constructs a subject string using JSAPI_CONSUMER_INFO format and sends a request using the makeRequestResponseRequired method from the same class. It expects a response message from the NATS server.

Then, it creates a ConsumerInfo object from the response message and checks if any error occurred during the retrieval process. If an error exists, it throws a JetStreamApiException. Otherwise, it returns the retrieved ConsumerInfo.

This method is part of the management functionality required by the regular context of the NatsJetStreamImpl class.

ConsumerInfo _createConsumer(String streamName, ConsumerConfiguration config) throws IOException, JetStreamApiException {
    // ConsumerConfiguration validates that name and durable are the same if both are supplied.
    String consumerName = config.getName();
    if (consumerName != null && !consumerCreate290Available) {
        throw JsConsumerCreate290NotAvailable.instance();
    }
    String durable = config.getDurable();
    String subj;
    if (consumerCreate290Available) {
        if (consumerName == null) {
            // if both consumerName and durable are null, generate a name
            consumerName = durable == null ? generateConsumerName() : durable;
        }
        String fs = config.getFilterSubject();
        if (fs == null || fs.equals(">")) {
            subj = String.format(JSAPI_CONSUMER_CREATE_V290, streamName, consumerName);
        } else {
            subj = String.format(JSAPI_CONSUMER_CREATE_V290_W_FILTER, streamName, consumerName, fs);
        }
    } else if (durable == null) {
        subj = String.format(JSAPI_CONSUMER_CREATE, streamName);
    } else {
        subj = String.format(JSAPI_DURABLE_CREATE, streamName, durable);
    }
    ConsumerCreateRequest ccr = new ConsumerCreateRequest(streamName, config);
    Message resp = makeRequestResponseRequired(subj, ccr.serialize(), conn.getOptions().getConnectionTimeout());
    return new ConsumerInfo(resp).throwOnHasError();
}

This method is defined in the class io.nats.client.impl.NatsJetStreamImpl and is responsible for creating a JetStream consumer with the given stream name and consumer configuration.

ConsumerInfo _createConsumer(String streamName, ConsumerConfiguration config) throws IOException, JetStreamApiException

1. Check if the consumerName in the config is not null and if consumerCreate290Available is false.

   • If true, throw an instance of JsConsumerCreate290NotAvailable, indicating that consumer creation is not available.

2. Get the durable option from the config.

3. Determine the subject for creating the consumer based on whether consumerCreate290Available is true or false.

4. If consumerCreate290Available is true, check if consumerName is null.

   • If both consumerName and durable are null, generate a consumer name using the generateConsumerName method.

5. Get the filterSubject from the config.

   • If filterSubject is null or equals to ">", format the subject using JSAPI_CONSUMER_CREATE_V290 with streamName and consumerName.
   • Otherwise, format the subject using JSAPI_CONSUMER_CREATE_V290_W_FILTER with streamName, consumerName, and filterSubject.

6. If consumerCreate290Available is false and durable is null, format the subject using JSAPI_CONSUMER_CREATE and streamName.

7. If none of the above conditions match, format the subject using JSAPI_DURABLE_CREATE with streamName and durable.

8. Create a ConsumerCreateRequest object with streamName and config.

9. Make a request with a response required using the subject and serialized ConsumerCreateRequest.

   • Use the connection timeout from the conn.getOptions().

10. Create a new ConsumerInfo object with the response message and throw an exception if the response has an error.

11. Return the ConsumerInfo object.

Note: The method throws IOException and JetStreamApiException.

The _createConsumer method is responsible for creating a consumer for a specific stream in the NATS JetStream implementation. It takes in the stream name and a ConsumerConfiguration object as parameters.

First, it validates that the consumer name is provided only if the consumerCreate290Available flag is false. If the flag is true and the consumer name is not provided, a name is generated.

Next, it determines the subject to use for creating the consumer based on the availability of the consumerCreate290Available flag and the presence of a filter subject.

Finally, it creates a ConsumerCreateRequest using the stream name and configuration, makes a request to create the consumer using the determined subject, and returns a ConsumerInfo object representing the created consumer. Any errors encountered during the creation process are thrown as JetStreamApiException.

void _createConsumerUnsubscribeOnException(String stream, ConsumerConfiguration cc, NatsJetStreamSubscription sub) throws IOException, JetStreamApiException {
    try {
        ConsumerInfo ci = _createConsumer(stream, cc);
        sub.setConsumerName(ci.getName());
    } catch (IOException | JetStreamApiException e) {
        // create consumer can fail, unsubscribe and then throw the exception to the user
        if (sub.getDispatcher() == null) {
            sub.unsubscribe();
        } else {
            sub.getDispatcher().unsubscribe(sub);
        }
        throw e;
    }
}

The _createConsumerUnsubscribeOnException method in the class NatsJetStreamImpl is responsible for creating a consumer for a specific stream using the given consumer configuration. If an exception occurs during the creation of the consumer, the method will automatically unsubscribe the subscription and then re-throw the exception to the user.

Here is a step-by-step description of what the method is doing:

1. The method accepts three parameters: stream (the name of the stream), cc (the consumer configuration), and sub (the NatsJetStreamSubscription object).
2. The method tries to create a consumer using the _createConsumer method, passing the stream and cc parameters. This method will return a ConsumerInfo object.
3. If the consumer creation is successful, the method sets the consumer name in the sub object using the setName method.
4. If an exception occurs during the consumer creation (either an IOException or a JetStreamApiException), the catch block is executed.
5. Inside the catch block, the method checks if the sub object has a dispatcher (a MessageDispatcher object) attached to it.
6. If the dispatcher is not present, the method calls the unsubscribe method on the sub object to unsubscribe it.
7. If the dispatcher is present, the method calls the unsubscribe method on the dispatcher, passing the sub object as a parameter.
8. Finally, the method re-throws the caught exception (e) to the user, ensuring that the calling code is aware of the failure that occurred during consumer creation.

By unsubscribing the subscription in case of an exception, the method ensures that no messages will be received by the consumer in an inconsistent state. Instead, the subscription is safely closed, and the exception is propagated to the user for appropriate error handling and reporting.

This method, _createConsumerUnsubscribeOnException, is used to create a consumer subscription for a given stream with the specified consumer configuration. It takes in three parameters: String stream, ConsumerConfiguration cc, and NatsJetStreamSubscription sub.

Inside the method, it first attempts to create the consumer using the _createConsumer method and assigns the consumer's name to the subscription. If any exception occurs during this process, the method catches it and performs the necessary cleanup.

If the subscription's dispatcher is null, it calls the unsubscribe() method to unsubscribe the subscription completely. Otherwise, it calls the unsubscribe(sub) method on the dispatcher.

Finally, regardless of whether an exception occurred or not, the method throws the caught exception to the user.

StreamInfo _getStreamInfo(String streamName, StreamInfoOptions options) throws IOException, JetStreamApiException {
    String subj = String.format(JSAPI_STREAM_INFO, streamName);
    StreamInfoReader sir = new StreamInfoReader();
    while (sir.hasMore()) {
        Message resp = makeRequestResponseRequired(subj, sir.nextJson(options), jso.getRequestTimeout());
        sir.process(resp);
    }
    return cacheStreamInfo(streamName, sir.getStreamInfo());
}

This method is defined in the class io.nats.client.impl.NatsJetStreamImpl. It takes in two parameters - streamName (a String) and options (a StreamInfoOptions object).

The method formats the subject using the streamName parameter and a constant JSAPI_STREAM_INFO. The formatted subject is stored in the subj variable.

A StreamInfoReader object sir is initialized. The method then enters a loop to read the StreamInfo response using the sir.hasMore() method.

Inside the loop, the method makes a request by calling the makeRequestResponseRequired method with the subj and sir.nextJson(options) as parameters. The jso.getRequestTimeout() value is passed as an argument as well.

The response received is then processed by calling the sir.process(resp) method.

After exiting the loop, the method caches the StreamInfo by calling the cacheStreamInfo method with the streamName and sir.getStreamInfo() as parameters.

Finally, it returns the cached StreamInfo as the output of the method.

Note: The method also throws two exceptions - IOException and JetStreamApiException.

The _getStreamInfo method in the NatsJetStreamImpl class is used to retrieve information about a stream in the NATS JetStream messaging system.

• The method takes two parameters: streamName (the name of the stream to get information about) and options (additional options for retrieving stream information).
• It constructs a subject string for sending a request to the server using the JSAPI_STREAM_INFO format.
• It creates a StreamInfoReader object to read the response messages from the server.
• It then enters a loop to process all the response messages and extract the stream information using the sir object.
• Finally, it caches the retrieved stream information using the cacheStreamInfo method and returns the cached information.

This method can throw an IOException if there is an error during the request-response process, and a JetStreamApiException if there is an error related to the JetStream API.

List<String> _getStreamNames(String subjectFilter) throws IOException, JetStreamApiException {
    StreamNamesReader snr = new StreamNamesReader();
    while (snr.hasMore()) {
        Message resp = makeRequestResponseRequired(JSAPI_STREAM_NAMES, snr.nextJson(subjectFilter), jso.getRequestTimeout());
        snr.process(resp);
    }
    return snr.getStrings();
}

This method is defined in the NatsJetStreamImpl class in the io.nats.client.impl package. It retrieves a list of stream names based on a subject filter.

• subjectFilter (Type: String): It is the filter used to match the stream names.

• IOException: This exception is thrown if there is an error while performing IO operations.
• JetStreamApiException: This exception is thrown if there is an error related to JetStream API.

1. Initialize a StreamNamesReader object named snr.
2. Enter a loop that continues while snr has more elements.
3. Inside the loop, perform the following steps:
   • Create a Message object resp by making a request with the JSAPI_STREAM_NAMES identifier, the nextJson obtained from snr using the subjectFilter, and the request timeout from the jso object.
   • Process the response resp using the snr.process method.
4. After the loop ends, return the strings obtained from the snr object using the getStrings method.

Please note that more implementation details might be present for some methods or variables used in this method, but they are not mentioned in the provided BODY.

The _getStreamNames method is a part of the NatsJetStreamImpl class in the io.nats.client.impl package. This method takes a subjectFilter as input and returns a list of stream names that match the filter.

The method starts by creating a StreamNamesReader object. It then enters a loop that continues as long as there are more stream names to process.

Within the loop, the method makes a request to retrieve the stream names by calling the makeRequestResponseRequired method. This method requires a response and throws an exception if it does not receive one. The request is made with the JSAPI_STREAM_NAMES parameter and the next JSON object from the StreamNamesReader which includes the subjectFilter and the request timeout from the jso object.

The response is then processed by calling the snr.process method, which updates the StreamNamesReader object with the names retrieved from the response.

Once there are no more stream names to process, the method returns the list of stream names by calling the snr.getStrings method.

This method may throw an IOException or JetStreamApiException, which should be handled by the calling code.

// General Utils // ---------------------------------------------------------------------------------------------------- ConsumerInfo lookupConsumerInfo(String streamName, String consumerName) throws IOException, JetStreamApiException

// ----------------------------------------------------------------------------------------------------
// General Utils
// ----------------------------------------------------------------------------------------------------
ConsumerInfo lookupConsumerInfo(String streamName, String consumerName) throws IOException, JetStreamApiException {
    try {
        return _getConsumerInfo(streamName, consumerName);
    } catch (JetStreamApiException e) {
        // the right side of this condition...  ( starting here \/ ) is for backward compatibility with server versions that did not provide api error codes
        if (e.getApiErrorCode() == JS_CONSUMER_NOT_FOUND_ERR || (e.getErrorCode() == 404 && e.getErrorDescription().contains("consumer"))) {
            return null;
        }
        throw e;
    }
}

The lookupConsumerInfo method in the NatsJetStreamImpl class is used to retrieve information about a specific consumer from a given stream.

Here is a step-by-step description of the method's functionality:

1. The method takes two parameters: streamName (the name of the stream) and consumerName (the name of the consumer).

2. The method tries to retrieve the consumer information by calling the _getConsumerInfo method with the provided streamName and consumerName.

3. If the _getConsumerInfo method throws a JetStreamApiException, the method checks the exception's apiErrorCode and errorCode.

4. If the apiErrorCode matches the JS_CONSUMER_NOT_FOUND_ERR constant or the errorCode is 404 and the error description contains the string "consumer", this means that the consumer was not found. In this case, the method returns null.

5. If the exception does not match the above conditions, it is re-thrown to be handled by the calling code.

Note: The method assumes that the _getConsumerInfo method has the necessary logic to retrieve the consumer information based on the provided stream and consumer names.

The lookupConsumerInfo method is used to retrieve information about a specific consumer in a JetStream stream. It takes in the streamName and consumerName as parameters and returns a ConsumerInfo object.

Internally, it calls the _getConsumerInfo method to retrieve the consumer information. If the consumer is not found, it handles backward compatibility by checking the error code and error description provided by the JetStream API. If the error code indicates a consumer not found error or the error description contains "consumer", the method returns null. Otherwise, it rethrows the JetStreamApiException for other types of errors.

// Request Utils // ---------------------------------------------------------------------------------------------------- Message makeRequestResponseRequired(String subject, byte[] bytes, Duration timeout) throws IOException

// ----------------------------------------------------------------------------------------------------
// Request Utils
// ----------------------------------------------------------------------------------------------------
Message makeRequestResponseRequired(String subject, byte[] bytes, Duration timeout) throws IOException {
    try {
        return responseRequired(conn.request(prependPrefix(subject), bytes, timeout));
    } catch (InterruptedException e) {
        throw new IOException(e);
    }
}

This method is used to send a request with a required response using the NATS messaging system. It takes in a subject, a byte array, and a timeout duration as inputs and returns a Message object.

1. Prepend the subject with the configured prefix using the prependPrefix method.
2. Send the request using the NATS connection object (conn) by calling the request method with the prepended subject, the byte array, and the timeout duration.
3. Call the responseRequired method to handle the response received from the request.
4. If the request is interrupted and an InterruptedException is thrown, wrap it in an IOException and throw it.

The makeRequestResponseRequired method is a utility method that is used to send a request message to a NATS server and waits for a response.

It takes three parameters:

• subject: The subject of the message.
• bytes: The message payload as a byte array.
• timeout: The duration for which the method will wait for a response before throwing an exception.

Internally, the method makes use of the conn.request method to send the request message to the NATS server. It prepends a prefix to the subject before sending the request.

If a response is received within the specified timeout duration, the method returns the received Message object. Otherwise, it throws an IOException.

Overall, makeRequestResponseRequired provides a convenient way to send request messages and handle the corresponding responses.

Message makeInternalRequestResponseRequired(String subject, Headers headers, byte[] data, Duration timeout, CancelAction cancelAction) throws IOException {
    try {
        return responseRequired(conn.requestInternal(subject, headers, data, timeout, cancelAction));
    } catch (InterruptedException e) {
        throw new IOException(e);
    }
}

The makeInternalRequestResponseRequired method is defined in the NatsJetStreamImpl class in the io.nats.client.impl package. The purpose of this method is to send an internal NATS request with the expectation of receiving a response.

Here is a step-by-step description of what this method is doing:

1. The method takes the following parameters:

   • subject (String): The subject of the NATS request.
   • headers (Headers): The headers to be included in the request.
   • data (byte[]): The payload data of the request.
   • timeout (Duration): The maximum duration to wait for a response.
   • cancelAction (CancelAction): An action to be executed if the request is cancelled.

2. The method tries to make the internal NATS request by calling the requestInternal method on the connection (conn) object. This method returns a Message object representing the response.

   • The requestInternal method is passed the subject, headers, data, timeout, and cancelAction parameters.
   • The method might throw an interrupted exception, so it is surrounded by a try-catch block.

3. If the requestInternal method call is successful, the method calls the responseRequired method to process the received response.

   • The responseRequired method is passed the Message object representing the response.
   • The result of this method call is returned by the makeInternalRequestResponseRequired method.

4. If an InterruptedException is caught during the execution of the try block, it is wrapped in an IOException and thrown.

In summary, the makeInternalRequestResponseRequired method sends an internal NATS request with the specified subject, headers, and payload data. It waits for a response up to the specified timeout duration, and then processes the received response before returning it.

The makeInternalRequestResponseRequired method is a part of the NatsJetStreamImpl class in the io.nats.client.impl package.

This method is responsible for making an internal request to the NATS server and requiring a response. It takes in the parameters subject (the subject of the message), headers (optional headers for the message), data (the payload of the message), timeout (the maximum time to wait for a response), and cancelAction (an action to execute if the request is cancelled).

Internally, this method calls the requestInternal method of the conn object (an instance of Connection) to make the request and waits for the response. It then returns the response as a Message object.

If the method is interrupted while waiting for the response, it throws an IOException.

Message responseRequired(Message respMessage) throws IOException {
    if (respMessage == null) {
        throw new IOException("Timeout or no response waiting for NATS JetStream server");
    }
    return respMessage;
}

The responseRequired method defined in the NatsJetStreamImpl class is responsible for handling the response message received from the NATS JetStream server. Here is a step-by-step description of what the method does based on its body:

1. The method takes a parameter respMessage, which is the response message received from the server.
2. If the respMessage is null, it means that the method has either timed out waiting for a response or there was no response from the NATS JetStream server.
3. In this case, the method throws an IOException with the message "Timeout or no response waiting for NATS JetStream server" to indicate the error condition.
4. If the respMessage is not null, it means that a response message was received from the server.
5. In this case, the method simply returns the respMessage as-is.

Overall, the responseRequired method allows you to handle the response message received from the NATS JetStream server, throwing an exception if no response was received or returning the response message if it was received successfully.

The responseRequired method is a part of the NatsJetStreamImpl class and it takes a Message object as input parameter and returns a Message object.

It checks if the input respMessage is null and throws an IOException with a message indicating a timeout or no response from the NATS JetStream server.

If the respMessage is not null, it simply returns the respMessage.

This method is used to verify if a response is required from the NATS JetStream server and handles the logic for handling timeouts or missing responses.

NatsJetStream

The NatsJetStream class is a subclass of NatsJetStreamImpl and implements the JetStream interface. It provides functionality for working with JetStream in NATS, a high-performance messaging system.

// Publish // ---------------------------------------------------------------------------------------------------- /** *

// ----------------------------------------------------------------------------------------------------
// Publish
// ----------------------------------------------------------------------------------------------------
/**
 * {@inheritDoc}
 */
@Override
public PublishAck publish(String subject, byte[] body) throws IOException, JetStreamApiException {
    return publishSyncInternal(subject, null, body, null);
}

The publish method defined in the NatsJetStream class is responsible for publishing a message to a subject in the NATS server using the NATS JetStream API.

Here is a step-by-step description of what the method is doing based on its body:

1. The method signature indicates that it returns a PublishAck object and can potentially throw IOException or JetStreamApiException.

2. The method overrides the publish method declared in a super class.

3. The subject and body parameters are passed to the method. The subject parameter specifies the subject to which the message will be published, and the body parameter is the byte array representing the content of the message.

4. The method calls the private method publishSyncInternal with the subject, null as the replyTo parameter, body, and null as the headers parameter.

5. The publishSyncInternal method is responsible for executing the actual publish operation synchronously.

6. The PublishAck object returned by publishSyncInternal is then returned by the publish method.

7. If an IOException or JetStreamApiException occurs during the execution, it is thrown by the method.

This method provides a convenient way to publish a message to a subject in NATS JetStream without specifying a reply subject or any headers.

The publish method in the NatsJetStream class is responsible for publishing a message to a specified subject in the NATS messaging system.

It takes two parameters: subject (which represents the subject/topic to publish the message to) and body (which is the actual content of the message in the form of a byte array).

Internally, it calls the publishSyncInternal method to publish the message synchronously, passing along the subject and body parameters. It also provides null values for two optional parameters: options and messageId.

The publish method then returns a PublishAck object, which represents an acknowledgment of the message publishing process.

In case of any exceptions or errors during the publishing process, such as IO errors or JetStream API exceptions, the method throws IOException or JetStreamApiException respectively, allowing the caller to handle the errors appropriately.

private PublishAck publishSyncInternal(String subject, Headers headers, byte[] data, PublishOptions options) throws IOException, JetStreamApiException {
    Headers merged = mergePublishOptions(headers, options);
    if (jso.isPublishNoAck()) {
        conn.publishInternal(subject, null, merged, data);
        return null;
    }
    Duration timeout = options == null ? jso.getRequestTimeout() : options.getStreamTimeout();
    Message resp = makeInternalRequestResponseRequired(subject, merged, data, timeout, CancelAction.COMPLETE);
    return processPublishResponse(resp, options);
}

The publishSyncInternal method is a function defined in the NatsJetStream class, located in the io.nats.client.impl package.

Here is a step-by-step description of what the method does:

1. The method accepts four parameters: subject (the subject to publish the message to), headers (the headers for the message), data (the message data as a byte array), and options (additional options for publishing).
2. The mergePublishOptions method is called to merge the headers and options parameters into a single Headers object named merged.
3. The method then checks if the PublishOptions object has the publishNoAck flag set. If it does, it calls the publishInternal method on the conn object (which represents the NATS connection) to publish the message without waiting for an acknowledgement. The subject, null for reply subject, merged for headers, and data are passed as parameters to the publishInternal method. The method then returns null since no acknowledgement is expected.
4. If the publishNoAck flag is not set, the method proceeds to get the request timeout value from the options object (or uses the default request timeout from jso if options is null) and assigns it to the timeout variable of type Duration.
5. The makeInternalRequestResponseRequired method is called with the subject, merged headers, data, timeout, and CancelAction.COMPLETE parameters. This method is responsible for making the internal request and waiting for a response. It returns a Message object named resp which represents the response received.
6. Finally, the processPublishResponse method is called, passing the resp object and the options object as parameters. This method is responsible for processing the publish response, including handling any errors, and returns a PublishAck object representing the acknowledgement of the publish operation. The PublishAck object is then returned by the publishSyncInternal method.

The method publishSyncInternal is responsible for publishing a message synchronously to NATS JetStream.

It takes in the following parameters:

• subject (String): The subject/topic to publish the message on.
• headers (Headers): The headers to be included with the message.
• data (byte[]): The actual data/message to be published.
• options (PublishOptions): Additional options for publishing.

First, the method merges the headers and options into a new header object called merged.

If the PublishOptions object's isPublishNoAck() method returns true, it means that the message should not expect an acknowledgement upon successful publishing. In this case, the publishInternal method is called on the conn object (which represents the NATS connection) to publish the message with the subject, null reply subject, merged headers, and data. The method then returns null.

If isPublishNoAck() returns false, it means that an acknowledgement is expected. The method then determines the timeout duration to wait for the response by checking the options object. If options is null, it uses the default request timeout from the jso (JetStreamOptions) object. Otherwise, it uses the StreamTimeout from the options object.

The makeInternalRequestResponseRequired method is called to make an internal request and wait for a response. It includes the subject, merged headers, data, timeout, and CancelAction.COMPLETE (which means the request is not cancellable). This method returns a Message object as the response.

Finally, the method processPublishResponse is called to process the resp (response) and return a PublishAck object, based on the provided options.

The method can throw IOException and JetStreamApiException, which indicates potential issues with the I/O operations or JetStream API calls.

private CompletableFuture<PublishAck> publishAsyncInternal(String subject, Headers headers, byte[] data, PublishOptions options, Duration knownTimeout) {
    Headers merged = mergePublishOptions(headers, options);
    if (jso.isPublishNoAck()) {
        conn.publishInternal(subject, null, merged, data);
        return null;
    }
    CompletableFuture<Message> future = conn.requestFutureInternal(subject, merged, data, knownTimeout, CancelAction.COMPLETE);
    return future.thenCompose(resp -> {
        try {
            responseRequired(resp);
            return CompletableFuture.completedFuture(processPublishResponse(resp, options));
        } catch (IOException | JetStreamApiException e) {
            throw new RuntimeException(e);
        }
    });
}

This method is defined in the class io.nats.client.impl.NatsJetStream and has the following signature:

private CompletableFuture<PublishAck> publishAsyncInternal(String subject, Headers headers, byte[] data, PublishOptions options, Duration knownTimeout)

The publishAsyncInternal method is responsible for asynchronously publishing a message to Nats JetStream.

1. Merge the headers and options to create a new Headers object named merged.
2. Check if the jso.isPublishNoAck() flag is set. If it is, perform the following steps:
   • Call the conn.publishInternal method to publish the message without acknowledgments. Pass the subject, null for the reply subject, merged headers, and the message data as arguments.
   • Return null indicating that no acknowledgement is expected.
3. If the jso.isPublishNoAck() flag is not set, proceed to the next step.
4. Call the conn.requestFutureInternal method to send a request message and get a CompletableFuture<Message> object named future. Pass the subject, merged headers, data, knownTimeout, and CancelAction.COMPLETE as arguments.
5. Use the future.thenCompose method to chain further processing after the response is received. Perform the following steps:
   • Inside the thenCompose callback function, check for any exceptions that occurred during the response processing. If an exception is caught, throw a RuntimeException with the caught exception as its cause.
   • If no exceptions occurred, call the responseRequired method to validate the response.
   • Call the processPublishResponse method to process the publish response and obtain a PublishAck object. Pass the resp (response) and options as arguments.
   • Wrap the PublishAck object in a completed CompletableFuture using CompletableFuture.completedFuture.
6. Return the CompletableFuture<PublishAck> object obtained from step 5.

Note: The method assumes that the necessary imports and declarations are present in the class.

The publishAsyncInternal method is used to asynchronously publish a message to a subject in NATS JetStream.

Here's a breakdown of what this method does:

1. It takes several parameters: subject (the subject to publish to), headers (optional headers of the message), data (the message payload as a byte array), options (optional publish options), and knownTimeout (timeout duration).

2. It merges the headers and options into a new Headers object called merged, using the mergePublishOptions method.

3. If the publish option isPublishNoAck is true, it calls the publishInternal method on the connection object (conn) to publish the message without waiting for an acknowledgement. It then returns null.

4. If the publish option isPublishNoAck is false, it creates a CompletableFuture called future by calling the requestFutureInternal method on the connection object (conn). This sends the message to the subject and expects a response from the server within the specified knownTimeout.

5. It then uses the thenCompose method on the future to handle the response from the server. This is where the logic for processing the response and handling any errors is implemented.

6. Inside the thenCompose block, it checks if a response is required (using the responseRequired method), and if so, it processes the response using the processPublishResponse method and returns a completed CompletableFuture with the result.

7. If any IOException or JetStreamApiException is caught during the processing of the response, it throws a RuntimeException.

In summary, the publishAsyncInternal method asynchronously publishes a message to a subject in NATS JetStream and handles the response from the server based on the specified publish options and timeout.

private PublishAck processPublishResponse(Message resp, PublishOptions options) throws IOException, JetStreamApiException {
    if (resp.isStatusMessage()) {
        throw new IOException("Error Publishing: " + resp.getStatus().getMessageWithCode());
    }
    PublishAck ack = new PublishAck(resp);
    String ackStream = ack.getStream();
    String pubStream = options == null ? null : options.getStream();
    // stream specified in options but different from ack should not happen but...
    if (pubStream != null && !pubStream.equals(ackStream)) {
        throw new IOException("Expected ack from stream " + pubStream + ", received from: " + ackStream);
    }
    return ack;
}

The processPublishResponse method in the NatsJetStream class is responsible for processing the response received after publishing a message. Here is a step-by-step description of what the method does:

1. The method takes two parameters: a Message object representing the response received and a PublishOptions object containing the options used for publishing the message.
2. The method first checks if the response is a status message. If it is, an IOException is thrown with an error message including the status message code and description.
3. If the response is not a status message, the method proceeds to create a new PublishAck object using the response. This PublishAck object represents the acknowledgment received for the published message.
4. The method then retrieves the stream name from the PublishAck object and assigns it to the variable ackStream.
5. Next, the method checks if the PublishOptions object is null. If it is, the variable pubStream is set to null.
6. If the PublishOptions object is not null, the method retrieves the stream name from the PublishOptions object and assigns it to the variable pubStream.
7. The method then checks if the pubStream is not null and if it is different from the ackStream. If they are different, an IOException is thrown with an error message indicating the expected stream from the pubStream variable and the actual stream received from the ackStream variable.
8. Finally, if none of the above conditions are met, the method returns the PublishAck object.

In summary, the processPublishResponse method handles the response received after publishing a message by checking for status messages, creating a PublishAck object, and performing stream comparison checks before returning the PublishAck object.

The processPublishResponse method in the NatsJetStream class is responsible for processing the response received after publishing a message.

The method takes two parameters - resp (representing the response message) and options (representing the options used for publishing). The method returns a PublishAck object.

If the response message indicates an error (status message), an IOException is thrown with the corresponding error message.

The method creates a PublishAck object using the response message, which represents the acknowledgment received for the published message.

It then checks if the stream specified in the options is different from the stream mentioned in the acknowledgment. If there is a mismatch, an IOException is thrown.

Finally, the method returns the PublishAck object.

Overall, this method handles error cases during publishing and ensures that the received acknowledgment matches the specified stream.

private Headers mergePublishOptions(Headers headers, PublishOptions opts) {
    // never touch the user's original headers
    Headers merged = headers == null ? null : new Headers(headers);
    if (opts != null) {
        merged = mergeNum(merged, EXPECTED_LAST_SEQ_HDR, opts.getExpectedLastSequence());
        merged = mergeNum(merged, EXPECTED_LAST_SUB_SEQ_HDR, opts.getExpectedLastSubjectSequence());
        merged = mergeString(merged, EXPECTED_LAST_MSG_ID_HDR, opts.getExpectedLastMsgId());
        merged = mergeString(merged, EXPECTED_STREAM_HDR, opts.getExpectedStream());
        merged = mergeString(merged, MSG_ID_HDR, opts.getMessageId());
    }
    return merged;
}

The mergePublishOptions method in the NatsJetStream class is used to merge the provided headers and PublishOptions object.

Here is a step-by-step description of what the method does:

1. It takes two parameters: headers and opts of types Headers and PublishOptions respectively.

2. It creates a new Headers object called merged by cloning the original headers provided as a parameter. If headers is null, merged will also be null.

3. It checks if the opts object is not null.

4. It calls the mergeNum method to merge the EXPECTED_LAST_SEQ_HDR (a constant value) header into merged with the value retrieved from opts.getExpectedLastSequence().

5. It calls the mergeNum method to merge the EXPECTED_LAST_SUB_SEQ_HDR (a constant value) header into merged with the value retrieved from opts.getExpectedLastSubjectSequence().

6. It calls the mergeString method to merge the EXPECTED_LAST_MSG_ID_HDR (a constant value) header into merged with the value retrieved from opts.getExpectedLastMsgId().

7. It calls the mergeString method to merge the EXPECTED_STREAM_HDR (a constant value) header into merged with the value retrieved from opts.getExpectedStream().

8. It calls the mergeString method to merge the MSG_ID_HDR (a constant value) header into merged with the value retrieved from opts.getMessageId().

9. It returns the merged Headers object.

Note: The mergeNum and mergeString methods are not defined in the provided code snippet. They are assumed to be helper methods that merge the provided key-value pair into the Headers object.

The method mergePublishOptions in the NatsJetStream class merges the given headers with the opts (PublishOptions) to create a new set of merged headers.

The original headers are never modified to ensure they remain intact. If headers is null, the merged headers will be null as well.

The method then proceeds to merge specific elements of opts into the merged headers using helper methods like mergeNum and mergeString. These elements include expectedLastSequence, expectedLastSubjectSequence, expectedLastMsgId, expectedStream, and messageId.

Finally, the merged headers are returned as the result of the method.

private Headers _mergeNum(Headers h, String key, String value) {
    if (h == null) {
        h = new Headers();
    }
    return h.add(key, value);
}

The _mergeNum method is a private method defined in the io.nats.client.impl.NatsJetStream class. This method takes three parameters:

1. Headers h: An instance of the Headers class, which represents the headers for a NatsJetStream request.
2. String key: The key of the header to be added or modified.
3. String value: The value of the header to be added or modified.

The purpose of the _mergeNum method is to merge a new header or modify an existing header in the Headers object. Here is a step-by-step description of what the method does:

1. Check if the h parameter is null.
2. If h is null, create a new instance of the Headers class and assign it to h.
3. Add or modify a header in the Headers object identified by the key parameter and set its value to the value parameter.
4. Return the modified Headers object.

The _mergeNum method ensures that a new header is added or an existing header is modified in the Headers object, which is then returned.

The _mergeNum method is a private method defined in the io.nats.client.impl.NatsJetStream class. This method takes three parameters: h, key, and value.

The purpose of this method is to merge the provided key-value pair into the given Headers object h. If the h object is null, a new Headers object is created. The method then adds the key-value pair to the Headers object using the add method.

Overall, the _mergeNum method ensures that the provided key-value pair is included in the Headers object, even if it is initially null, and returns the updated Headers object.

JetStreamSubscription createSubscription(String subject, String queueName, NatsDispatcher dispatcher, MessageHandler userHandler, boolean isAutoAck, PushSubscribeOptions pushSubscribeOptions, PullSubscribeOptions pullSubscribeOptions) throws IOException, JetStreamApiException {
    // 1. Prepare for all the validation
    boolean isPullMode = pullSubscribeOptions != null;
    SubscribeOptions so;
    String stream;
    String qgroup;
    ConsumerConfiguration userCC;
    if (isPullMode) {
        // options must have already been checked to be non-null
        so = pullSubscribeOptions;
        stream = pullSubscribeOptions.getStream();
        userCC = so.getConsumerConfiguration();
        // just to make compiler happy both paths set variable
        qgroup = null;
        validateNotSupplied(userCC.getDeliverGroup(), JsSubPullCantHaveDeliverGroup);
        validateNotSupplied(userCC.getDeliverSubject(), JsSubPullCantHaveDeliverSubject);
    } else {
        so = pushSubscribeOptions == null ? DEFAULT_PUSH_OPTS : pushSubscribeOptions;
        // might be null, that's ok (see directBind)
        stream = so.getStream();
        userCC = so.getConsumerConfiguration();
        if (userCC.maxPullWaitingWasSet()) {
            throw JsSubPushCantHaveMaxPullWaiting.instance();
        }
        if (userCC.maxBatchWasSet()) {
            throw JsSubPushCantHaveMaxBatch.instance();
        }
        if (userCC.maxBytesWasSet()) {
            throw JsSubPushCantHaveMaxBytes.instance();
        }
        // figure out the queue name
        qgroup = validateMustMatchIfBothSupplied(userCC.getDeliverGroup(), queueName, JsSubQueueDeliverGroupMismatch);
        if (so.isOrdered() && qgroup != null) {
            throw JsSubOrderedNotAllowOnQueues.instance();
        }
        if (dispatcher != null && (so.getPendingMessageLimit() != Consumer.DEFAULT_MAX_MESSAGES || so.getPendingByteLimit() != Consumer.DEFAULT_MAX_BYTES)) {
            throw JsSubPushAsyncCantSetPending.instance();
        }
    }
    // 2A. Flow Control / heartbeat not always valid
    if (userCC.getIdleHeartbeat() != null && userCC.getIdleHeartbeat().toMillis() > 0) {
        if (isPullMode) {
            throw JsSubFcHbNotValidPull.instance();
        }
        if (qgroup != null) {
            throw JsSubFcHbNotValidQueue.instance();
        }
    }
    // 2B. Did they tell me what stream? No? look it up.
    final String fnlStream;
    if (stream == null) {
        fnlStream = lookupStreamBySubject(subject);
        if (fnlStream == null) {
            throw JsSubNoMatchingStreamForSubject.instance();
        }
    } else {
        fnlStream = stream;
    }
    ConsumerConfiguration serverCC = null;
    String consumerName = userCC.getDurable();
    if (consumerName == null) {
        consumerName = userCC.getName();
    }
    String inboxDeliver = userCC.getDeliverSubject();
    // 3. Does this consumer already exist?
    if (consumerName != null) {
        ConsumerInfo serverInfo = lookupConsumerInfo(fnlStream, consumerName);
        if (serverInfo != null) {
            // the consumer for that durable already exists
            serverCC = serverInfo.getConsumerConfiguration();
            // check to see if the user sent a different version than the server has
            // because modifications are not allowed during create subscription
            ConsumerConfigurationComparer userCCC = new ConsumerConfigurationComparer(userCC);
            List<String> changes = userCCC.getChanges(serverCC);
            if (changes.size() > 0) {
                throw JsSubExistingConsumerCannotBeModified.instance("Changed fields: " + changes);
            }
            // deliver subject must be null/empty for pull, defined for push
            if (isPullMode) {
                if (!nullOrEmpty(serverCC.getDeliverSubject())) {
                    throw JsSubConsumerAlreadyConfiguredAsPush.instance();
                }
            } else if (nullOrEmpty(serverCC.getDeliverSubject())) {
                throw JsSubConsumerAlreadyConfiguredAsPull.instance();
            }
            if (serverCC.getDeliverGroup() == null) {
                // lookedUp was null, means existing consumer is not a queue consumer
                if (qgroup == null) {
                    // ok fine, no queue requested and the existing consumer is also not a queue consumer
                    // we must check if the consumer is in use though
                    if (serverInfo.isPushBound()) {
                        throw JsSubConsumerAlreadyBound.instance();
                    }
                } else {
                    // else they requested a queue but this durable was not configured as queue
                    throw JsSubExistingConsumerNotQueue.instance();
                }
            } else if (qgroup == null) {
                throw JsSubExistingConsumerIsQueue.instance();
            } else if (!serverCC.getDeliverGroup().equals(qgroup)) {
                throw JsSubExistingQueueDoesNotMatchRequestedQueue.instance();
            }
            // durable already exists, make sure the filter subject matches
            if (nullOrEmpty(subject)) {
                // allowed if they had given both stream and durable
                subject = userCC.getFilterSubject();
            } else if (!isFilterMatch(subject, serverCC.getFilterSubject(), fnlStream)) {
                throw JsSubSubjectDoesNotMatchFilter.instance();
            }
            // use the deliver subject as the inbox. It may be null, that's ok, we'll fix that later
            inboxDeliver = serverCC.getDeliverSubject();
        } else if (so.isBind()) {
            throw JsSubConsumerNotFoundRequiredInBind.instance();
        }
    }
    // 4. If pull or no deliver subject (inbox) provided or found, make an inbox.
    final String fnlInboxDeliver;
    if (isPullMode) {
        fnlInboxDeliver = conn.createInbox() + ".*";
    } else if (inboxDeliver == null) {
        fnlInboxDeliver = conn.createInbox();
    } else {
        fnlInboxDeliver = inboxDeliver;
    }
    // 5. If consumer does not exist, create and settle on the config. Name will have to wait
    //    If the consumer exists, I know what the settled info is
    final String settledConsumerName;
    final ConsumerConfiguration settledServerCC;
    if (serverCC == null) {
        ConsumerConfiguration.Builder ccBuilder = ConsumerConfiguration.builder(userCC);
        // Pull mode doesn't maintain a deliver subject. It's actually an error if we send it.
        if (!isPullMode) {
            ccBuilder.deliverSubject(fnlInboxDeliver);
        }
        if (userCC.getFilterSubject() == null) {
            ccBuilder.filterSubject(subject);
        }
        ccBuilder.deliverGroup(qgroup);
        settledServerCC = ccBuilder.build();
        settledConsumerName = null;
    } else {
        settledServerCC = serverCC;
        settledConsumerName = consumerName;
    }
    // 6. create the subscription. lambda needs final or effectively final vars
    final MessageManager mm;
    final NatsSubscriptionFactory subFactory;
    if (isPullMode) {
        MessageManagerFactory mmFactory = so.isOrdered() ? _pullOrderedMessageManagerFactory : _pullMessageManagerFactory;
        mm = mmFactory.createMessageManager(conn, this, fnlStream, so, settledServerCC, false, dispatcher == null);
        subFactory = (sid, lSubject, lQgroup, lConn, lDispatcher) -> new NatsJetStreamPullSubscription(sid, lSubject, lConn, lDispatcher, this, fnlStream, settledConsumerName, mm);
    } else {
        MessageManagerFactory mmFactory = so.isOrdered() ? _pushOrderedMessageManagerFactory : _pushMessageManagerFactory;
        mm = mmFactory.createMessageManager(conn, this, fnlStream, so, settledServerCC, false, dispatcher == null);
        subFactory = (sid, lSubject, lQgroup, lConn, lDispatcher) -> {
            NatsJetStreamSubscription nsub = new NatsJetStreamSubscription(sid, lSubject, lQgroup, lConn, lDispatcher, this, fnlStream, settledConsumerName, mm);
            if (lDispatcher == null) {
                nsub.setPendingLimits(so.getPendingMessageLimit(), so.getPendingByteLimit());
            }
            return nsub;
        };
    }
    NatsJetStreamSubscription sub;
    if (dispatcher == null) {
        sub = (NatsJetStreamSubscription) conn.createSubscription(fnlInboxDeliver, qgroup, null, subFactory);
    } else {
        AsyncMessageHandler handler = new AsyncMessageHandler(mm, userHandler, isAutoAck, settledServerCC);
        sub = (NatsJetStreamSubscription) dispatcher.subscribeImplJetStream(fnlInboxDeliver, qgroup, handler, subFactory);
    }
    // 7. The consumer might need to be created, do it here
    if (settledConsumerName == null) {
        _createConsumerUnsubscribeOnException(fnlStream, settledServerCC, sub);
    }
    return sub;
}

This method is defined in the io.nats.client.impl.NatsJetStream class. It is used to create a JetStream subscription. The method takes various parameters and returns a JetStreamSubscription.

JetStreamSubscription createSubscription(String subject, String queueName, NatsDispatcher dispatcher, MessageHandler userHandler, boolean isAutoAck, PushSubscribeOptions pushSubscribeOptions, PullSubscribeOptions pullSubscribeOptions) throws IOException, JetStreamApiException

• subject: The subject for the subscription.
• queueName: The name of the queue (optional).
• dispatcher: The message dispatcher for the subscription (optional).
• userHandler: The user-defined message handler.
• isAutoAck: A boolean flag indicating whether the messages should be auto-acknowledged.
• pushSubscribeOptions: Options for push subscriptions (optional).
• pullSubscribeOptions: Options for pull subscriptions (optional).

1. Prepare for validation

   • Check if the subscription is in pull mode or push mode.
   • Set the appropriate options and configurations based on the mode.
   • Validate that certain options are not supplied in pull mode.
   • Validate that certain options are not supplied in push mode.

2. Flow Control / Heartbeat validation

   • If an idle heartbeat is set and greater than zero:
     • Throw an exception if the subscription is in pull mode.
     • Throw an exception if a consumer group is specified.

3. Stream lookup

   • If the stream is not provided:
     • Look up the stream based on the subject.
     • Throw an exception if no matching stream is found.

4. Consumer existence check

   • If a consumer name is provided:
     • Look up the consumer information in the server.
     • If the consumer exists:
       • Get the server-side consumer configuration.
       • Compare the user configuration with the server configuration.
       • Throw an exception if there are any changes in the configurations.
       • Validate that the deliver subject is appropriate for the subscription mode.
       • Validate that the consumer is not already bound or configured as a push/pull consumer.
     • If the consumer does not exist and binding is required:
       • Throw an exception indicating that the consumer is not found.

5. Inbox creation

   • If the subscription is in pull mode or no deliver subject is provided:
     • Create an inbox for message delivery.
   • Otherwise, use the provided deliver subject.

6. Consumer configuration and creation

   • If the consumer does not exist:
     • Build the consumer configuration based on the user configuration.
     • Set the deliver subject if not in pull mode.
     • Set the filter subject if it is not already set.
     • Set the deliver group.
   • Otherwise, use the settled server-side configuration and consumer name.
   • Create the message manager based on the subscription mode and other parameters.
   • Set the subscription factory based on the mode and message manager.
   • Create the JetStream subscription based on the provided parameters and subscription factory.
   • Set the pending limits if the message dispatcher is not provided.
   • Create and associate the asynchronous message handler if the dispatcher is provided.
   • Call the appropriate subscription creation method based on the presence of the dispatcher.

7. Consumer creation

   • If the consumer was not found, create the consumer on the server and handle unsubscribe on exception.

8. Return the created subscription

Note: The method throws IOException and JetStreamApiException in case of any errors.

The createSubscription method in the NatsJetStream class is used to create a subscription for consuming messages from a specified subject in JetStream.

Here is a high-level overview of what the method does:

1. It performs validation based on the subscription options provided, such as whether it's a push or pull subscription and if any conflicting options are present.
2. It checks if the consumer configuration already exists, and if it does, verifies that the configuration matches the provided options.
3. If the consumer does not exist or if it is a new consumer, it creates a settled configuration based on the provided options.
4. It determines the deliver subject for the subscription, either by creating a new inbox or using the existing deliver subject.
5. Based on whether it is a push or pull subscription, it creates the appropriate type of subscription with the necessary message manager implementation.
6. If a dispatcher is provided, it subscribes to the deliver subject using the dispatcher's subscribeImplJetStream method. Otherwise, it directly creates a subscription using the connection's createSubscription method.
7. If a new consumer was created, it calls the _createConsumerUnsubscribeOnException method to handle any errors and unsubscribe from the subscription in case of an exception.
8. Finally, it returns the created subscription object.

Please note that this is a high-level description and the method may have more intricate details and mechanisms not covered in this summary.

private boolean isFilterMatch(String subscribeSubject, String filterSubject, String stream) throws IOException, JetStreamApiException {
    // subscribeSubject guaranteed to not be null
    // filterSubject may be null or empty or have value
    if (subscribeSubject.equals(filterSubject)) {
        return true;
    }
    if (nullOrEmpty(filterSubject) || filterSubject.equals(">")) {
        // lookup stream subject returns null if there is not exactly one subject
        String streamSubject = lookupStreamSubject(stream);
        return subscribeSubject.equals(streamSubject);
    }
    return false;
}

This method is used to determine if a filter subject matches with a given subscribe subject. It takes three parameters - subscribeSubject, filterSubject, and stream.

1. Check if subscribeSubject is equal to filterSubject:

   • If true, return true (indicating a match).
   • If false, proceed to the next step.

2. Check if filterSubject is either null, empty, or equals to >.

   • If true, perform a lookup of the stream subject by calling the lookupStreamSubject method with the stream parameter.
     • If the lookup returns null (indicating there is not exactly one subject), return false (indicating there is no match).
     • If the lookup returns a non-null value, compare it with subscribeSubject.
       • If they are equal, return true (indicating a match).
       • If they are not equal, return false (indicating no match).
   • If false, return false (indicating no match).

3. Return false as a default fallback (indicating no match).

The isFilterMatch method is a private method defined in the NatsJetStream class. It takes three parameters: subscribeSubject, filterSubject, and stream. The method checks whether the subscribeSubject is a match for the filterSubject based on certain conditions.

First, it checks if the subscribeSubject is equal to the filterSubject. If they are equal, it returns true.

Next, it checks if the filterSubject is null, empty, or has the value ">". If any of these conditions is true, it calls the lookupStreamSubject method to get the streamSubject and then checks if the subscribeSubject is equal to the streamSubject. If they are equal, it returns true.

If none of the above conditions are met, it returns false.

This method is used to determine whether a given subject should be filtered based on the subscription subject and the filter subject in the context of the NatsJetStream implementation.

public JetStreamSubscription subscribe(String subject, String queue, PushSubscribeOptions options) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, String queue, PushSubscribeOptions options) throws IOException, JetStreamApiException {
    validateSubject(subject, isSubjectRequired(options));
    queue = emptyAsNull(validateQueueName(queue, false));
    return createSubscription(subject, queue, null, null, false, options, null);
}

The subscribe method in the NatsJetStream class is used to create a subscription to a subject in the NATS JetStream server.

Here is a step-by-step description of what the method does based on its code:

1. Validates the subject parameter to ensure it is not null or empty, and checks if it is required based on the options parameter.
2. Validates the queue parameter to ensure it is not empty and converts any empty strings to null if allowed.
3. Calls the createSubscription method with the following parameters:
   • subject: The subject to subscribe to.
   • queue: The queue to receive messages from (can be null).
   • null (placeholder): Placeholder parameter for the durableName parameter that is not used in this method.
   • null (placeholder): Placeholder parameter for the deliverStartTime parameter that is not used in this method.
   • false: The manualAcks parameter that specifies whether to manually acknowledge messages.
   • options: The PushSubscribeOptions object that contains additional options for the subscription.
   • null (placeholder): Placeholder parameter for the flowControl parameter that is not used in this method.
4. Returns the JetStreamSubscription object representing the subscription.

This method handles exceptions such as IOException and JetStreamApiException that can be thrown during the subscription process.

The subscribe method in the NatsJetStream class is used to create a subscription for consuming messages from a NATS JetStream.

This method takes three parameters: subject (the subject to subscribe to), queue (an optional queue group name for load balancing), and options (options for configuring the subscription).

Firstly, the method validates the subject and checks if it is required based on the options. If the queue parameter is empty, it is set to null. Then, the method calls the createSubscription method, passing the subject, queue, and other necessary parameters to create the subscription. Finally, the method returns a JetStreamSubscription object representing the created subscription.

If any exceptions occur during the execution of this method, such as an IOException or a JetStreamApiException, they are thrown.

public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, boolean autoAck) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, boolean autoAck) throws IOException, JetStreamApiException {
    validateSubject(subject, true);
    validateNotNull(dispatcher, "Dispatcher");
    validateNotNull(handler, "Handler");
    return createSubscription(subject, null, (NatsDispatcher) dispatcher, handler, autoAck, null, null);
}

The subscribe method in the NatsJetStream class is used to create a JetStream subscription on a specified subject. It takes in the following parameters:

1. subject (String): The subject to subscribe to.
2. dispatcher (Dispatcher): An instance of a dispatcher responsible for delivering messages to the handler.
3. handler (MessageHandler): An instance of the message handler responsible for processing incoming messages.
4. autoAck (boolean): A flag indicating whether messages should be automatically acknowledged.

The method performs the following steps:

1. Validate the subject to ensure it is not null or empty. If it fails validation, throw an IllegalArgumentException.
2. Validate the dispatcher to ensure it is not null. If it fails validation, throw an IllegalArgumentException.
3. Validate the message handler to ensure it is not null. If it fails validation, throw an IllegalArgumentException.
4. Create a new subscription by invoking the createSubscription method and passing the subject, null, the dispatcher casted to NatsDispatcher, the handler, the autoAck flag, and two null values as arguments.
5. Return the newly created subscription.

Note: The implementation of createSubscription is not provided in the given snippet, so the exact details of how the subscription is created are not available.

The subscribe method is used to create a subscription for consuming messages from a subject in the NATS JetStream messaging system.

It takes the following parameters:

• subject: The subject to subscribe to.
• dispatcher: A dispatcher object responsible for dispatching incoming messages to the appropriate handler.
• handler: A message handler object that processes incoming messages.
• autoAck: A boolean flag indicating whether the subscription should automatically acknowledge messages after they have been successfully processed.

The method first performs some validation checks on the subject, dispatcher, and handler parameters to ensure they are not null. Then, it calls the createSubscription method internally to create the actual subscription object and returns it.

Overall, the method provides a convenient way to create a subscription for consuming messages from a subject in the NATS JetStream system, by specifying the subject, dispatcher, handler, and auto-acknowledgment settings.

public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, boolean autoAck, PushSubscribeOptions options) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, boolean autoAck, PushSubscribeOptions options) throws IOException, JetStreamApiException {
    validateSubject(subject, isSubjectRequired(options));
    validateNotNull(dispatcher, "Dispatcher");
    validateNotNull(handler, "Handler");
    return createSubscription(subject, null, (NatsDispatcher) dispatcher, handler, autoAck, options, null);
}

This method is defined in the io.nats.client.impl.NatsJetStream class. It takes in the following parameters:

• subject (String): The subject to subscribe to.
• dispatcher (Dispatcher): The dispatcher to use for message handling.
• handler (MessageHandler): The handler for processing received messages.
• autoAck (boolean): Flag indicating whether messages should be automatically acknowledged.
• options (PushSubscribeOptions): Optional options for configuring the subscription.

The method performs the following steps:

1. Validate the subject parameter by calling the validateSubject method, passing in the subject and the result of the isSubjectRequired method called with the options parameter. This ensures that the subject is valid and meets any configured requirements.

2. Validate that the dispatcher parameter is not null by calling the validateNotNull method with the dispatcher and the string "Dispatcher" as arguments. This ensures that a valid dispatcher is provided for message handling.

3. Validate that the handler parameter is not null by calling the validateNotNull method with the handler and the string "Handler" as arguments. This ensures that a valid message handler is provided for processing received messages.

4. Call the createSubscription method, passing in the subject, null for the queueGroup parameter, the dispatcher cast to NatsDispatcher, the handler, the autoAck flag, the options, and null for the jsSub parameter. This method creates and returns a JetStreamSubscription object based on the provided parameters.

5. Return the JetStreamSubscription object created in the previous step.

Note: This method can throw an IOException or a JetStreamApiException, which may occur during the validation or subscription creation process.

The subscribe() method in the NatsJetStream class is used to subscribe to a subject in NATS JetStream.

It takes in the following parameters:

• subject (String): The subject to subscribe to.
• dispatcher (Dispatcher): An object that dispatches incoming messages to the appropriate handler.
• handler (MessageHandler): An object that handles incoming messages.
• autoAck (boolean): A flag indicating whether the subscription should automatically acknowledge the messages.
• options (PushSubscribeOptions): Additional options for the subscription.

The method first validates the subject and the dispatcher to ensure they are not null. It then creates a subscription using the createSubscription() method, passing in the subject, null for the interest, the dispatcher, the handler, autoAck flag, options, and null for the flow control options.

The method returns a JetStreamSubscription object representing the subscription.

public JetStreamSubscription subscribe(String subject, String queue, Dispatcher dispatcher, MessageHandler handler, boolean autoAck, PushSubscribeOptions options) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, String queue, Dispatcher dispatcher, MessageHandler handler, boolean autoAck, PushSubscribeOptions options) throws IOException, JetStreamApiException {
    validateSubject(subject, isSubjectRequired(options));
    queue = emptyAsNull(validateQueueName(queue, false));
    validateNotNull(dispatcher, "Dispatcher");
    validateNotNull(handler, "Handler");
    return createSubscription(subject, queue, (NatsDispatcher) dispatcher, handler, autoAck, options, null);
}

The subscribe method in the NatsJetStream class is used to create a JetStream subscription for consuming messages from a subject. Here is a step-by-step description of what the method does based on its body:

1. Validates the subject parameter to ensure it is not null and meets the requirements specified by the options parameter.
2. Validates the queue parameter, which represents the name of the consumer group queue, to ensure it is not empty and returns null if it is empty.
3. Validates the dispatcher parameter, which is responsible for dispatching messages to the handler, to ensure it is not null.
4. Validates the handler parameter, which is responsible for handling incoming messages, to ensure it is not null.
5. Calls the createSubscription method with the validated parameters to create a JetStream subscription.
6. Returns the created JetStream subscription.

The subscribe method throws an IOException if there is an error during the subscription creation process and a JetStreamApiException if there is an error with the JetStream API.

The subscribe method in the io.nats.client.impl.NatsJetStream class is used to create a subscription to a JetStream in NATS.

It takes the following parameters:

• subject: The subject to subscribe to.
• queue: The optional queue group name for the subscription.
• dispatcher: The dispatcher used for message dispatching.
• handler: The handler for processing received messages.
• autoAck: A flag indicating if the messages should be automatically acknowledged.
• options: Additional options for the subscription.

The method begins by validating the subject and queue parameters. It ensures that the subject is not empty and checks if the queue name is provided or not.

Next, it checks that both the dispatcher and handler objects are not null, as they are required to process the received messages.

Finally, it calls the createSubscription method with the validated parameters to actually create the subscription and returns the resulting JetStreamSubscription object.

If any errors occur during the subscription creation, such as IO exceptions or JetStream API exceptions, they are caught and thrown.

public JetStreamSubscription subscribe(String subject, PullSubscribeOptions options) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, PullSubscribeOptions options) throws IOException, JetStreamApiException {
    validateSubject(subject, isSubjectRequired(options));
    validateNotNull(options, "Pull Subscribe Options");
    return createSubscription(subject, null, null, null, false, null, options);
}

The subscribe method in the NatsJetStream class is responsible for creating a JetStream subscription. Here is a step-by-step breakdown of what the method does based on the provided body:

1. The method overrides the subscribe method from the parent class.
2. The method takes two parameters: subject (the subject to subscribe to) and options (the pull subscription options).
3. Before proceeding, the method validates the subject by calling the validateSubject method, passing the subject and a check if the subject is required based on the options.
4. The method then validates that the options are not null, using the validateNotNull method, passing the options and a descriptive error message.
5. Finally, the method calls the createSubscription method, passing various parameters:
   • The subject parameter provided to the subscribe method.
   • null for the Queue parameter.
   • null for the Durable parameter.
   • null for the DeliverSubject parameter.
   • false for the NoAck parameter.
   • null for the AckWait parameter.
   • The options parameter provided to the subscribe method.
6. The method returns the result of the createSubscription method, which is a JetStreamSubscription.

Overall, the subscribe method validates the subject and options parameters, and then calls another method to create and return a JetStream subscription based on the provided parameters.

The subscribe method in the NatsJetStream class, defined in io.nats.client.impl, is used to create a subscription to consume messages from a JetStream subject using pull-based consumption.

The method takes two parameters:

• subject: The subject (topic) to subscribe to.
• options: The pull subscription options, which customize the behavior of the subscription.

The method first performs some input validation, ensuring that the subject is valid and that the options are not null. If the validation passes, it then calls the createSubscription method internally to create and return a JetStreamSubscription object.

The returned JetStreamSubscription is the subscription that can be used to receive messages from the specified subject using pull-based consumption.

Possible exceptions that can be thrown by the method include IOException (in case of an I/O error) and JetStreamApiException (if there is an error related to JetStream API).

Note that the provided code only shows the method body, and the implementation of the createSubscription method is not provided.

public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, PullSubscribeOptions options) throws IOException, JetStreamApiException

@Override
public JetStreamSubscription subscribe(String subject, Dispatcher dispatcher, MessageHandler handler, PullSubscribeOptions options) throws IOException, JetStreamApiException {
    validateSubject(subject, isSubjectRequired(options));
    validateNotNull(dispatcher, "Dispatcher");
    validateNotNull(handler, "Handler");
    validateNotNull(options, "Pull Subscribe Options");
    return createSubscription(subject, null, (NatsDispatcher) dispatcher, handler, false, null, options);
}

The subscribe method is defined in the class io.nats.client.impl.NatsJetStream. It takes four parameters: subject, dispatcher, handler, and options.

1. The method first calls the validateSubject method to ensure that the subject parameter is valid. It checks if the subject is required based on the options parameter by calling the isSubjectRequired method.
2. It then calls the validateNotNull method for the dispatcher, handler, and options parameters to ensure that they are not null.
3. Finally, it calls the createSubscription method to create a JetStream subscription. It passes the subject, null, the dispatcher casted to NatsDispatcher, the handler and the options parameters to the createSubscription method. The last two parameters in the createSubscription method call are false and null.

The method returns a JetStreamSubscription object obtained from the createSubscription method call.

The subscribe method in class NatsJetStream is used to create a subscription to a given subject in the NATS JetStream messaging system.

It takes in the following parameters:

• subject (string): The subject to subscribe to.
• dispatcher (Dispatcher): The dispatcher to use for delivering messages.
• handler (MessageHandler): The handler to process incoming messages.
• options (PullSubscribeOptions): The options for the subscription.

This method first performs some validations on the subject, dispatcher, handler, and options to ensure they are not null.

Then, it calls the createSubscription method to create the subscription. It passes the subject, dispatcher, handler, and options to the createSubscription method, along with some additional parameters.

Finally, it returns the created JetStreamSubscription object.

public ConsumerContext consumerContext(String streamName, String consumerName) throws IOException, JetStreamApiException

@Override
public ConsumerContext consumerContext(String streamName, String consumerName) throws IOException, JetStreamApiException {
    Validator.validateStreamName(streamName, true);
    Validator.required(consumerName, "Consumer Name");
    return getNatsStreamContext(streamName).consumerContext(consumerName);
}

The consumerContext method in the NatsJetStream class is used to retrieve a ConsumerContext object based on the provided streamName and consumerName. Here is a step-by-step description of what this method does:

1. The method is declared with the @Override annotation, indicating that it overrides a method in a superclass or interface.

2. The validateStreamName method from the Validator class is called with the streamName as an argument to check if it is valid. The true boolean value indicates that an exception should be thrown if the stream name is invalid.

3. The required method from the Validator class is called with the consumerName and a string value "Consumer Name" as arguments to check if the consumer name is not null or empty. If it is null or empty, an exception is thrown.

4. The getNatsStreamContext method is called with the streamName as an argument to retrieve the corresponding NatsStreamContext object.

5. The consumerContext method is called on the NatsStreamContext object returned in the previous step with the consumerName as an argument. This method retrieves the ConsumerContext object associated with the specified consumer name.

6. The retrieved ConsumerContext object is returned as the result of the method.

Note: The method includes throws declarations for IOException and JetStreamApiException, indicating that it can throw these exceptions during its execution.

The consumerContext method is a part of the NatsJetStream class in the io.nats.client.impl package. It takes two parameters streamName and consumerName, both of type String.

The method first validates the streamName parameter using the validateStreamName method from the Validator class, ensuring that it is not empty or null.

Next, it checks that the consumerName parameter is not empty or null, using the required method from the Validator class.

Finally, it calls the consumerContext method on the NatsStreamContext object obtained from the getNatsStreamContext method. This method returns a ConsumerContext object, which is then returned by the consumerContext method.

Overall, this method is responsible for validating the input parameters and returning the consumer context for the specified stream and consumer.

StreamInfoReader

The StreamInfoReader class is responsible for reading information about a stream. It provides methods to retrieve various details such as the stream's name, duration, and format. This class acts as a utility for obtaining relevant information about a stream in a convenient manner.

void process(Message msg) throws JetStreamApiException {
    engine = new ListRequestEngine(msg);
    StreamInfo si = new StreamInfo(msg);
    if (streamInfo == null) {
        streamInfo = si;
    } else {
        streamInfo.getStreamState().getSubjects().addAll(si.getStreamState().getSubjects());
    }
}

This method is defined in the class io.nats.client.impl.StreamInfoReader. It is responsible for processing a message and updating the stream information.

• msg: The message to be processed.

1. Create a new ListRequestEngine object named engine by passing the msg as a parameter.
2. Create a new StreamInfo object named si by passing the msg as a parameter.
3. Check if the variable streamInfo is null.
   • If it is null, assign the value of si to streamInfo.
   • If it is not null, add all the subjects from si.getStreamState().getSubjects() to streamInfo.getStreamState().getSubjects().

The process method in the io.nats.client.impl.StreamInfoReader class is responsible for processing a message received by the reader.

1. First, a new ListRequestEngine is created using the received message.
2. Then, a StreamInfo object is created based on the received message.
3. If the streamInfo object is currently null, the streamInfo variable is set to the newly created si object.
4. If the streamInfo object is not null, the subjects from the si object's stream state are added to the subjects of the existing streamInfo object.

In summary, the process method processes a message by creating a ListRequestEngine and a StreamInfo object, and updates the streamInfo object with the received message's stream state subjects.

byte[] nextJson(StreamInfoOptions options) {
    StringBuilder sb = beginJson();
    addField(sb, "offset", engine.nextOffset());
    if (options != null) {
        addField(sb, SUBJECTS_FILTER, options.getSubjectsFilter());
        addFldWhenTrue(sb, DELETED_DETAILS, options.isDeletedDetails());
    }
    return endJson(sb).toString().getBytes();
}

The nextJson method in the StreamInfoReader class is used to return the JSON representation of the next stream information entry based on the given options.

Here is a step-by-step description of what the method does:

1. Initialize a StringBuilder named sb by calling the beginJson method.
2. Add the "offset" field to the sb with the value obtained from calling the nextOffset method on the engine object.
3. Check if the options parameter is not null.
4. If options is not null:
   • Add the "subjects_filter" field to the sb with the value obtained from calling the getSubjectsFilter method on the options object.
   • Check if the isDeletedDetails method on the options object returns true.
   • If it does, add the "deleted_details" field to the sb.
5. Call the endJson method on the sb to finalize the JSON representation.
6. Convert the sb to a String and then get the corresponding byte representation as a byte[].
7. Return the byte array.

The nextJson method in the StreamInfoReader class is responsible for generating a JSON representation of the next stream information.

Here is a brief description of what this method does:

• Creates a StringBuilder to build the JSON string.
• Adds the "offset" field to the JSON string by calling the nextOffset method of the underlying engine.
• Checks if the options parameter is not null:
  • If not null, adds the "subjectsFilter" field to the JSON string by calling the getSubjectsFilter method of the options parameter.
  • Adds the "deletedDetails" field to the JSON string by calling the isDeletedDetails method of the options parameter.
• Returns the JSON string representation as a byte array.

This method is used to get the next JSON representation of the stream information, including offset, subjects filter, and deleted details if specified by the options parameter.

FileAuthHandler

The FileAuthHandler class is an implementation of the AuthHandler interface. It provides functionality for handling authentication in a file-based system.

private char[] extract(CharBuffer data, int headers) {
    CharBuffer buff = CharBuffer.allocate(data.length());
    boolean skipLine = false;
    int headerCount = 0;
    int linePos = -1;
    while (data.length() > 0) {
        char c = data.get();
        linePos++;
        // End of line, either we got it, or we should keep reading the new line
        if (c == '\n' || c == '\r') {
            if (buff.position() > 0) {
                // we wrote something
                break;
            }
            skipLine = false;
            // so we can start right up
            linePos = -1;
            continue;
        }
        // skip to the new line
        if (skipLine) {
            continue;
        }
        // Ignore whitespace
        if (Character.isWhitespace(c)) {
            continue;
        }
        // If we are on a - skip that line, bump the header count
        if (c == '-' && linePos == 0) {
            skipLine = true;
            headerCount++;
            continue;
        }
        // Skip the line, or add to buff
        if (!skipLine && headerCount == headers) {
            buff.put(c);
        }
    }
    // check for naked value
    if (buff.position() == 0 && headers == 1) {
        data.position(0);
        while (data.length() > 0) {
            char c = data.get();
            if (c == '\n' || c == '\r' || Character.isWhitespace(c)) {
                if (buff.position() > 0) {
                    // we wrote something
                    break;
                }
                continue;
            }
            buff.put(c);
        }
        buff.flip();
    } else {
        buff.flip();
    }
    char[] retVal = new char[buff.length()];
    buff.get(retVal);
    buff.clear();
    for (int i = 0; i < buff.capacity(); i++) {
        buff.put('\0');
    }
    return retVal;
}
```### private char[] readKeyChars() throws IOException 
```java
private char[] readKeyChars() throws IOException {
    char[] keyChars = null;
    if (this.credsFile != null) {
        byte[] data = Files.readAllBytes(Paths.get(this.credsFile));
        ByteBuffer bb = ByteBuffer.wrap(data);
        CharBuffer chars = StandardCharsets.UTF_8.decode(bb);
        // we are 2nd so 3 headers
        keyChars = this.extract(chars, 3);
        // Clear things up as best we can
        chars.clear();
        for (int i = 0; i < chars.capacity(); i++) {
            chars.put('\0');
        }
        for (int i = 0; i < data.length; i++) {
            data[i] = 0;
        }
    } else {
        byte[] data = Files.readAllBytes(Paths.get(this.nkeyFile));
        ByteBuffer bb = ByteBuffer.wrap(data);
        CharBuffer chars = StandardCharsets.UTF_8.decode(bb);
        keyChars = this.extract(chars, 1);
        // Clear things up as best we can
        chars.clear();
        for (int i = 0; i < chars.capacity(); i++) {
            chars.put('\0');
        }
        for (int i = 0; i < data.length; i++) {
            data[i] = 0;
        }
    }
    return keyChars;
}
```### public byte[] sign(byte[] nonce) 
```java
public byte[] sign(byte[] nonce) {
    try {
        char[] keyChars = this.readKeyChars();
        NKey nkey = NKey.fromSeed(keyChars);
        byte[] sig = nkey.sign(nonce);
        nkey.clear();
        return sig;
    } catch (Exception exp) {
        throw new IllegalStateException("problem signing nonce", exp);
    }
}

The sign method in the class io.nats.client.impl.FileAuthHandler is used to sign a given byte array using a secret key stored in a file.

Here is a step-by-step description of what the method does:

1. Reads the secret key stored in the file as characters and stores them in a char array called keyChars.

2. Converts the keyChars array into an NKey object called nkey using the fromSeed method of the NKey class. The fromSeed method takes the characters of the secret key and creates a NKey object, which represents a cryptographic key derived from the seed.

3. The nkey.sign(nonce) method is called to sign the given nonce byte array using the key stored in the nkey object. This method internally uses a cryptographic algorithm to generate a digital signature based on the key and the nonce.

4. The nkey.clear() method is called to clear the internal state of the nkey object. This is done to remove any sensitive information from memory after the signing process.

5. The generated signature is returned as a byte array.

6. If any exception occurs during the signing process, an IllegalStateException is thrown with the message "problem signing nonce". The original exception is wrapped and included as the cause of the IllegalStateException.

The sign method in the FileAuthHandler class is responsible for signing the provided nonce bytes using a private key stored in a file.

First, the method reads the private key characters from the file using the readKeyChars method.

Then, it creates an NKey object from the private key characters.

Next, it uses the sign method of the NKey object to sign the nonce bytes.

After signing, it clears the private key from memory using the clear method of the NKey object.

Finally, it returns the signature as a byte array.

If any exception occurs during the signing process, it throws an IllegalStateException with an appropriate error message.

NatsSubscription

The NatsSubscription class extends the NatsConsumer class and implements the Subscription interface. It represents a subscription to a NATS messaging system. This class provides functionality to subscribe to topics and receive messages from the NATS server. Additionally, it inherits properties and methods from the NatsConsumer class, allowing for efficient message consumption and processing.

void reSubscribe(String newDeliverSubject) {
    connection.sendUnsub(this, 0);
    if (dispatcher == null) {
        connection.remove(this);
        sid = connection.reSubscribe(this, newDeliverSubject, queueName);
    } else {
        MessageHandler handler = dispatcher.getSubscriptionHandlers().get(sid);
        dispatcher.remove(this);
        sid = dispatcher.reSubscribe(this, newDeliverSubject, queueName, handler);
    }
    subject = newDeliverSubject;
}

The reSubscribe method in the NatsSubscription class is responsible for resubscribing the subscription to a new deliver subject.

Here is a step-by-step description of what the method does:

1. The method starts by sending an unsubscribe request using the sendUnsub method of the connection associated with the subscription. This unsubscribes the subscription from the current subject it was subscribed to.

2. After unsubscribing, the method checks if the subscription has a dispatcher assigned to it. The dispatcher is responsible for handling incoming messages for the subscription.

3. If the subscription does not have a dispatcher, it means it is a single-subscription and not part of a subscription group. In this case, the subscription is removed from the connection using the remove method.

4. The method then proceeds to re-subscribe the subscription to the new deliver subject using the reSubscribe method of the connection. This method takes the subscription, the new deliver subject, and the existing queue name (if any) as parameters. The reSubscribe method returns the new subscription ID (sid).

5. If the subscription has a dispatcher assigned to it, it means it is part of a subscription group. In this case, the method retrieves the message handler associated with the subscription using the getSubscriptionHandlers method of the dispatcher.

6. The method then removes the subscription from the dispatcher using the remove method.

7. Finally, the method re-subscribes the subscription to the new deliver subject using the reSubscribe method of the dispatcher. This method takes the subscription, the new deliver subject, the existing queue name (if any), and the message handler as parameters. The reSubscribe method returns the new subscription ID (sid).

8. The method updates the subscription's subject with the new deliver subject.

That concludes the step-by-step description of the reSubscribe method.

The reSubscribe method in the NatsSubscription class is used to resubscribe the subscription to a new deliver subject.

It first sends an unsubscribe message to the NATS server for the current subscription using the connection.sendUnsub method.

If the dispatcher object is null, it means that the subscription is standalone and not part of a dispatcher. In this case, it removes the subscription from the connection's list of active subscriptions using connection.remove, and then resubscribes the subscription to the new deliver subject by calling connection.reSubscribe method with the new deliver subject and the existing queue name.

If the dispatcher object is not null, it means that the subscription is part of a dispatcher. In this case, it looks up the message handler associated with the subscription using dispatcher.getSubscriptionHandlers().get(sid). Then it removes the subscription from the dispatcher's list of active subscriptions using dispatcher.remove, and finally resubscribes the subscription to the new deliver subject by calling dispatcher.reSubscribe with the new deliver subject, the existing queue name, and the retrieved message handler.

After the resubscription, the subject instance variable is updated with the new deliver subject.

void invalidate() {
    if (this.incoming != null) {
        this.incoming.pause();
    }
    this.dispatcher = null;
    this.incoming = null;
}

This method is defined in the class io.nats.client.impl.NatsSubscription and is responsible for invalidating the subscription.

1. Check if the incoming object is not null:

   • If incoming is not null, proceed to the next step.
   • If incoming is null, skip to step 4.

2. Call the pause() method on the incoming object:

   • incoming.pause() is called to pause the incoming messages on the subscription.

3. Set both the dispatcher and incoming objects to null:

   • dispatcher is set to null so that no further dispatching of messages can occur.
   • incoming is set to null so that the subscription no longer holds a reference to incoming messages.

4. The method execution is completed.

The invalidate method in the NatsSubscription class, defined in the io.nats.client.impl package, is used to mark the subscription as invalid.

When called, the method checks if the incoming field is not null. If it is not null, it pauses the incoming object, which is most likely some kind of message or data receiver for the subscription.

Next, it sets both the dispatcher and incoming fields to null thereby invalidating the subscription. This means that any further messages or data for this subscription will be ignored.

Overall, the invalidate method is responsible for pausing incoming messages and invalidating the subscription by setting relevant fields to null.

protected NatsMessage nextMessageInternal(Duration timeout) throws InterruptedException {
    if (this.dispatcher != null) {
        throw new IllegalStateException("Subscriptions that belong to a dispatcher cannot respond to nextMessage directly.");
    } else if (this.incoming == null) {
        throw new IllegalStateException("This subscription is inactive.");
    }
    NatsMessage msg = incoming.pop(timeout);
    if (this.incoming == null || !this.incoming.isRunning()) {
        // We were unsubscribed while waiting
        throw new IllegalStateException("This subscription became inactive.");
    }
    if (msg != null) {
        this.incrementDeliveredCount();
    }
    if (this.reachedUnsubLimit()) {
        this.connection.invalidate(this);
    }
    return msg;
}

The method nextMessageInternal is part of the class io.nats.client.impl.NatsSubscription and is used to retrieve the next message from the subscription's incoming queue. Here is a step-by-step description of what this method does:

1. Check if the subscription has a dispatcher assigned to it. If it does, it means that the subscription belongs to a dispatcher and cannot respond to nextMessage directly. If this condition is true, an IllegalStateException is thrown.

2. Check if the incoming queue is null. If it is, it means that this subscription is inactive and cannot retrieve the next message. If this condition is true, an IllegalStateException is thrown.

3. Use the pop method of the incoming queue to retrieve the next message. The pop method blocks until a message is available or until the specified timeout duration has passed.

4. Check if the incoming queue is null or not running anymore. If it is null or not running, it means that the subscription was unsubscribed while waiting for the next message. If this condition is true, an IllegalStateException is thrown.

5. If a message was retrieved successfully (i.e., msg is not null), increment the delivered count of the subscription.

6. Check if the subscription has reached the unsubscribe limit. If it has, it means that the subscription should be invalidated by the connection. The connection's invalidate method is called with the current subscription as a parameter.

7. Finally, return the retrieved message (msg) to the caller.

Note: The timeout parameter is used to specify the maximum time to wait for a message. It is of type Duration, which represents a duration-based amount of time. If no message is available within the specified timeout, the pop method will return null.

The nextMessageInternal method in the NatsSubscription class is used to retrieve the next message that has been received by the subscription.

Here's a breakdown of what the method does:

1. It first checks if the subscription belongs to a dispatcher. If it does, an IllegalStateException is thrown because subscriptions belonging to a dispatcher cannot respond to nextMessage directly.

2. It then checks if the subscription is active. If it is not, an IllegalStateException is thrown indicating that the subscription is inactive.

3. The method then waits for the next message to be received, using the specified timeout duration.

4. After waiting for a message, it checks if the subscription is still active. If it has become inactive during the waiting period, an IllegalStateException is thrown.

5. If a message is received, the method increments the delivered count of the subscription.

6. It then checks if the subscription has reached the unsubscribe limit. If it has, the associated connection is invalidated.

7. Finally, the method returns the received message.

Overall, the nextMessageInternal method ensures that the subscription is active, waits for the next message, handles subscription and connection state changes, and returns the received message.

public void unsubscribe()

@Override
public void unsubscribe() {
    if (this.dispatcher != null) {
        throw new IllegalStateException("Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly.");
    } else if (this.incoming == null) {
        throw new IllegalStateException("This subscription is inactive.");
    }
    if (isDraining()) {
        // No op while draining
        return;
    }
    this.connection.unsubscribe(this, -1);
}

The unsubscribe method in the class io.nats.client.impl.NatsSubscription is responsible for unsubscribing a subscription from a NATS server.

Here's a step-by-step description of what this method does based on its body:

1. Check if the dispatcher object of the subscription is not null. If it is not null, it means that the subscription belongs to a dispatcher, and it is not allowed to respond to unsubscribe directly. In such a case, throw an IllegalStateException with the message "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly."

2. If the incoming object of the subscription is null, it means that the subscription is inactive and cannot be unsubscribed. In this case, throw an IllegalStateException with the message "This subscription is inactive."

3. Check if the subscription is currently in the process of draining messages. If it is, then there is no need to perform any action, so return without doing anything.

4. If none of the above conditions are met, call the unsubscribe method of the connection object (which represents the NATS server connection) with the subscription object and -1 as parameters. This effectively unsubscribes the subscription from the NATS server.

That's the step-by-step description of what the unsubscribe method does based on its body.

The unsubscribe method in the NatsSubscription class is used to unsubscribe from a NATS subscription.

Here's a breakdown of what the method does:

1. It first checks if the subscription is associated with a dispatcher. If it is, it throws an IllegalStateException with the message that subscriptions belonging to a dispatcher cannot directly respond to unsubscribe requests.

2. It then checks if the subscription is inactive by checking if the incoming field is null. If it is, it throws an IllegalStateException with the message that the subscription is inactive.

3. Next, it checks if the subscription is currently being drained. If it is, it simply returns and does nothing. This implies that the unsubscription request is disregarded while the subscription is being drained.

4. If none of the above conditions are met, it proceeds to call the unsubscribe method on the underlying NATS connection, passing in the subscription instance and a timeout value of -1, indicating an immediate unsubscribe request.

public Subscription unsubscribe(int after)

@Override
public Subscription unsubscribe(int after) {
    if (this.dispatcher != null) {
        throw new IllegalStateException("Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly.");
    } else if (this.incoming == null) {
        throw new IllegalStateException("This subscription is inactive.");
    }
    if (isDraining()) {
        // No op while draining
        return this;
    }
    this.connection.unsubscribe(this, after);
    return this;
}

The unsubscribe method in the NatsSubscription class is used to unsubscribe from a NATS subscription.

Step-by-step description of the method:

1. Check if the subscription has a dispatcher assigned to it:

   • If a dispatcher is assigned, it means that the subscription belongs to a dispatcher and cannot directly respond to an unsubscribe. In this case, an IllegalStateException is thrown with the error message "Subscriptions that belong to a dispatcher cannot respond to unsubscribe directly."

2. Check if the subscription is inactive:

   • If the incoming field is null, it means that the subscription is inactive and cannot be unsubscribed. In this case, an IllegalStateException is thrown with the error message "This subscription is inactive."

3. Check if the subscription is currently draining:

   • If the subscription is in the process of draining, meaning that it is in the process of shutting down, no additional unsubscribe operations can be performed. In this case, the method returns the current instance of the NatsSubscription object without performing any further action.

4. If none of the above conditions are met, perform the unsubscribe operation:

   • The connection object associated with the subscription is used to perform the actual unsubscribe operation by invoking the unsubscribe method on it. The after parameter specified in the method call determines the number of server replies to wait for before considering the unsubscribe successful.

5. Finally, return the current instance of the NatsSubscription object after performing the unsubscribe operation.

Please note that this description is based on the provided code snippet and may not capture all the possible scenarios or interactions with other parts of the codebase.

The unsubscribe method is used to unsubscribe from a NATS subscription. It takes an integer parameter after, which specifies the number of messages to allow before automatically unsubscribing.

If the subscription belongs to a dispatcher, it throws an IllegalStateException since subscriptions that belong to a dispatcher cannot directly respond to unsubscribe.

If the subscription is inactive (i.e., there is no incoming message), it also throws an IllegalStateException.

If the subscription is currently being drained (i.e., actively unsubscribing), the method returns itself (this) without performing any operation.

Otherwise, it calls the unsubscribe method on the connection associated with the subscription, passing in the subscription instance and the after parameter. Lastly, it returns itself (this).

NatsServerPool

The NatsServerPool class is a public implementation of the ServerPool interface.

public void initialize(Options opts) {
    // 1. Hold on to options as we need them for settings
    options = opts;
    // 2. maxConnectAttempts accounts for the first connect attempt and also reconnect attempts
    maxConnectAttempts = options.getMaxReconnect() < 0 ? Integer.MAX_VALUE : options.getMaxReconnect() + 1;
    // 3. Add all the bootstrap to the server list and prepare list for next
    //    FYI bootstrap will always have at least the default url
    synchronized (listLock) {
        entryList = new ArrayList<>();
        for (NatsUri nuri : options.getNatsServerUris()) {
            // 1. If item is not found in the list being built, add to the list
            boolean notAlreadyInList = true;
            for (ServerPoolEntry entry : entryList) {
                if (nuri.equivalent(entry.nuri)) {
                    notAlreadyInList = false;
                    break;
                }
            }
            if (notAlreadyInList) {
                if (defaultScheme == null && !nuri.getScheme().equals(NatsConstants.NATS_PROTOCOL)) {
                    defaultScheme = nuri.getScheme();
                }
                entryList.add(new ServerPoolEntry(nuri, false));
            }
        }
        // 6. prepare list for next
        afterListChanged();
    }
}

The initialize method in the NatsServerPool class is responsible for setting up the server pool based on the provided options.

1. Hold on to options as we need them for settings:

   • The options variable is assigned the value of the opts parameter, which holds the options for configuring the server pool.

2. Calculate the maximum number of connect attempts:

   • The maxConnectAttempts variable is assigned a value based on the maxReconnect option from the options object.
   • If maxReconnect is less than 0, maxConnectAttempts is set to Integer.MAX_VALUE, indicating an infinite number of reconnect attempts.
   • Otherwise, maxConnectAttempts is set to maxReconnect + 1.

3. Add all the bootstrap to the server list:

   • A synchronized block is used to ensure thread safety when modifying the entryList variable.
   • The entryList variable is initialized as a new ArrayList.
   • For each NatsUri in the natsServerUris option from the options object:
     1. Check if the item is not already in the list being built.
     2. If it is not already in the list, create a new ServerPoolEntry with the NatsUri and add it to the entryList.
     3. Additionally, if the defaultScheme is null and the current NatsUri has a scheme different from the NatsConstants.NATS_PROTOCOL, assign the scheme of the NatsUri to the defaultScheme.

4. Prepare list for the next operation:

   • Call the afterListChanged method to handle any necessary updates or adjustments after the entryList has been modified.

This method initializes the server pool by storing the options, determining the maximum number of connect attempts, building the initial list of server entries based on the provided URIs, and performing any required preparations for subsequent operations.

The initialize method in the NatsServerPool class initializes the NATS server pool using the provided options. Here is a breakdown of what the method does:

1. Stores the provided options internally for later use.
2. Calculates the maximum number of allowed connection attempts based on the maximum reconnect value from the options.
3. Populates the server list by adding all the bootstrap servers specified in the options. If a server is already in the list, it won't be added again.
4. If a defaultScheme hasn't been set yet and the first server in the list has a non-default scheme, the defaultScheme is set to that scheme.
5. Each server in the list is wrapped in a ServerPoolEntry object and added to the entryList.
6. Once the server list has been updated, any necessary operations for handling the change in the list are performed.

public boolean acceptDiscoveredUrls(List discoveredServers)

@Override
public boolean acceptDiscoveredUrls(List<String> discoveredServers) {
    // 1. If ignored discovered servers, don't do anything b/c never want
    //    anything but the explicit, which is already loaded.
    // 2. return false == no new servers discovered
    if (options.isIgnoreDiscoveredServers()) {
        return false;
    }
    synchronized (listLock) {
        // 2. Build a list for discovered
        //    - since we'll need the NatsUris later
        //    - and to have a list to use to prune removed gossiped servers
        List<NatsUri> discovered = new ArrayList<>();
        for (String d : discoveredServers) {
            try {
                discovered.add(new NatsUri(d, defaultScheme));
            } catch (URISyntaxException ignore) {
                // should never actually happen
            }
        }
        // 3. Start a new server list, loading in current order from the current list, and keeping
        //    - the last connected
        //    - all non-gossiped
        //    - any found in the new discovered list
        //      - for any new discovered, we also remove them from
        //        that list so step there are no dupes for step #4
        //      - This also maintains the Srv state of an already known discovered
        List<ServerPoolEntry> newEntryList = new ArrayList<>();
        for (ServerPoolEntry entry : entryList) {
            int ix = findEquivalent(discovered, entry.nuri);
            if (ix != -1 || entry.nuri.equals(lastConnected) || !entry.isGossiped) {
                newEntryList.add(entry);
                if (ix != -1) {
                    discovered.remove(ix);
                }
            }
        }
        // 4. Add all left over from the new discovered list
        boolean discoveryContainedUnknowns = false;
        if (discovered.size() > 0) {
            discoveryContainedUnknowns = true;
            for (NatsUri d : discovered) {
                newEntryList.add(new ServerPoolEntry(d, true));
            }
        }
        // 5. replace the list with the new one
        entryList = newEntryList;
        // 6. prepare list for next
        afterListChanged();
        // 7.
        return discoveryContainedUnknowns;
    }
}

The acceptDiscoveredUrls method in the NatsServerPool class is responsible for updating the server pool based on the newly discovered NATS server URLs.

Here is a step-by-step description of what the method does:

1. It first checks if the option to ignore discovered servers is enabled. If it is, the method returns false, indicating that no new servers have been discovered.

2. If the ignore option is not enabled, the method proceeds to update the server pool.

3. Inside a synchronized block, the method creates a new list called discovered, which will be used to store the newly discovered server URIs as NatsUri objects (using the defaultScheme).

4. For each URL in the discoveredServers list, the method attempts to create a new NatsUri object. If the URL is invalid and results in a URISyntaxException, the exception is ignored.

5. After creating the discovered list, the method starts building a new list called newEntryList, which will replace the existing server pool list.

6. For each entry in the existing server pool list (entryList), the method checks if the entry matches any of the discovered server URIs or the last connected URI, or if the entry is not gossiped.

7. If a match is found or the entry is not gossiped, the entry is added to the newEntryList. If a match is found in the discovered list (discovered), that entry is removed from the discovered list to prevent duplicates in the next step.

8. After iterating through all entries in the existing server pool list, the method checks if there are any leftover entries in the discovered list. If there are, it means that these entries were not matched with any existing entries and are considered newly discovered servers.

9. For each remaining entry in the discovered list, a new ServerPoolEntry is created and added to the newEntryList.

10. After adding all the necessary entries to the newEntryList, the method replaces the existing server pool list (entryList) with the newEntryList.

11. The afterListChanged method is then called to perform any necessary operations after the server pool list has been updated.

12. Finally, the method returns a boolean value, indicating whether any unknown servers were discovered during the process. This value is determined by checking if the size of the discovered list is greater than zero.

That's the step-by-step description of what the acceptDiscoveredUrls method does based on its body.

The acceptDiscoveredUrls method is used to update the list of servers in the NATS server pool based on new discovered URLs.

Here's a step-by-step breakdown of what the method does:

1. If the option to ignore discovered servers is enabled, the method does nothing and returns false.

2. If the ignore option is not enabled, the method proceeds and creates a synchronized block.

3. Within the synchronized block, the method builds a new list of NatsUri objects based on the discovered server URLs provided as input.

4. It then creates a new list called newEntryList to store the updated server pool entries. It iterates over the existing entryList and adds the entries that match the following conditions:

   • The entry has an equivalent NatsUri in the discovered list.
   • The entry is the last connected server.
   • The entry is a non-gossiped server.

5. After adding the relevant entries from the existing list, the method checks if there are any remaining NatsUri objects in the discovered list. If there are, it iterates over them and adds new ServerPoolEntry objects to the newEntryList with the isGossiped flag set to true.

6. The entryList is then replaced with the updated newEntryList.

7. The afterListChanged() method is called to perform any necessary actions after the server list has been updated.

8. Finally, the method returns true if the discovered list contained any unknown server URLs, indicating that new servers were added to the server pool. Otherwise, it returns false.

Overall, the acceptDiscoveredUrls method is responsible for accepting new discovered server URLs and updating the server pool accordingly.

private void afterListChanged() {
    // 1. randomize if needed and allowed
    if (entryList.size() > 1 && !options.isNoRandomize()) {
        Collections.shuffle(entryList, ThreadLocalRandom.current());
    }
    // 2. calculate hasSecureServer and find the index of lastConnected
    hasSecureServer = false;
    int lastConnectedIx = -1;
    for (int ix = 0; ix < entryList.size(); ix++) {
        NatsUri nuri = entryList.get(ix).nuri;
        hasSecureServer |= nuri.isSecure();
        if (nuri.equals(lastConnected)) {
            lastConnectedIx = ix;
        }
    }
    // C. put the last connected server at the end of the list
    if (lastConnectedIx != -1) {
        entryList.add(entryList.remove(lastConnectedIx));
    }
}

The afterListChanged method in the NatsServerPool class performs the following steps:

1. Check if the entryList has more than one element and if the noRandomize option is not set. If both conditions are met, randomize the order of the elements in the entryList using Collections.shuffle method and a ThreadLocalRandom generator.

2. Calculate the hasSecureServer flag by iterating through each element in the entryList. Get the nuri (NatsUri) from the element and set hasSecureServer to true if any of the nuri is secure (isSecure() method returns true). Additionally, find the index of the lastConnected NatsUri in the entryList and store it in the lastConnectedIx variable.

3. Reorder the entryList such that the element at the lastConnectedIx position is moved to the end of the list. This is done using the add and remove methods of entryList, ensuring that the last connected server is now at the end of the list.

The method afterListChanged is responsible for performing some operations after the list of server entries is changed. Here's a breakdown of what the method does:

1. Randomize: If the entryList has more than one element and the options do not specify to not randomize, the method shuffles the elements in the entryList using Collections.shuffle() with the help of ThreadLocalRandom.current(). This step ensures that the order of the servers is randomized.

2. Calculate hasSecureServer and find the index of lastConnected: The method iterates through each element in the entryList and checks if the server URL (NatsUri) is secure. It sets the hasSecureServer flag to true if any of the server URLs are secure. It also finds the index of the last connected server by comparing the server's URL with the lastConnected variable.

3. Put the last connected server at the end of the list: If a last connected server was found (i.e., lastConnectedIx is not -1), the method removes the element at lastConnectedIx from the entryList and adds it back at the end of the list. This step ensures that the last connected server is always at the end of the list.

Overall, this method is responsible for randomizing the server list, checking for secure servers, and ensuring the last connected server is moved to the end of the list.

public NatsUri peekNextServer()

@Override
public NatsUri peekNextServer() {
    synchronized (listLock) {
        return entryList.size() > 0 ? entryList.get(0).nuri : null;
    }
}

The peekNextServer method in the io.nats.client.impl.NatsServerPool class is used to retrieve the next server in a list of server entries. Below is a step-by-step description of what the method does:

1. The method peekNextServer is implemented and overrides the same method from the parent class or interface.
2. The method is declared to return an object of type NatsUri.
3. The method is synchronized using the synchronized keyword, which means only one thread can access this code block at a time.
4. Inside the synchronized block, the method checks if the entryList (a list of server entries) has a size greater than 0.
5. If the entryList has at least one entry, the method returns the nuri (a field of the first entry in the entryList).
6. If the entryList is empty, i.e., it doesn't have any entries, the method returns null.

The peekNextServer method in the NatsServerPool class returns the URI of the next server in the pool without removing it from the list. It achieves this by synchronizing access to the list of server entries and checking if the list is not empty. If it is not empty, it retrieves the first entry's URI from the list and returns it. Otherwise, it returns null.

public NatsUri nextServer()

@Override
public NatsUri nextServer() {
    // 0. The list is already managed for qualified by connectFailed
    // 1. Get the first item in the list, update it's time, add back to the end of list
    synchronized (listLock) {
        if (entryList.size() > 0) {
            ServerPoolEntry entry = entryList.remove(0);
            entry.lastAttempt = System.currentTimeMillis();
            entryList.add(entry);
            return entry.nuri;
        }
        return null;
    }
}

The nextServer() method is used to retrieve the next available server from the list of server entries in the NatsServerPool class. Here is a step-by-step description of what the method is doing:

1. It first checks if the size of the entry list is greater than 0 to ensure there are available servers.
2. If there are available servers, it retrieves the first server entry from the list using the remove(0) method, which removes and returns the element at index 0.
3. It then updates the last attempt time of the server entry using entry.lastAttempt = System.currentTimeMillis(), which sets the last attempt time to the current system time in milliseconds.
4. After updating the last attempt time, it adds the server entry back to the end of the list using entryList.add(entry).
5. Finally, it returns the NatsUri (URI of the server) from the retrieved server entry using return entry.nuri.

If there are no available servers (i.e., the entry list is empty), it returns null.

Note: The code uses synchronization with the listLock object to ensure thread safety when accessing and modifying the entry list.

The nextServer method in the NatsServerPool class is used to retrieve the next available NatsUri from the server pool. Here's what the method does:

1. It checks if the entryList (which represents the list of available servers) is not empty.
2. If the entryList is not empty, it removes the first server entry (ServerPoolEntry) from the list.
3. It updates the lastAttempt timestamp of the removed server entry with the current system time in milliseconds.
4. It adds the updated server entry back to the end of the entryList.
5. Finally, it returns the NatsUri associated with the removed server entry.