Benchmarking io.lettuce.core.protocol.CommandHandler.decode() method

LettuceByteBufBenchmark.java

import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.ExecutionException;

public class LettuceByteBufBenchmark {

    private static <T> CompletionStage<List<T>> join(List<CompletionStage<T>> stages) {
        CompletionStage<List<T>> current = CompletableFuture.completedFuture(new ArrayList<T>());
        for (int i = 0; i < stages.size(); ++i) {
            current = current.thenCombine(stages.get(i), (List<T> acc, T val) -> {
                acc.add(val);
                return acc;
            });
        }
        return current;
    }

    private static void prepareData(
            RedisAsyncCommands<String, String> commands, int numOfCommands) throws ExecutionException,
            InterruptedException {
        commands.setAutoFlushCommands(false);
        List<CompletionStage<String>> responseFutures = new ArrayList<>(numOfCommands);
        for (int i = 0; i < numOfCommands; i++) {
            responseFutures.add(commands.set("key-" + i, "value-" + i));
        }
        commands.flushCommands();
        commands.setAutoFlushCommands(true);
        join(responseFutures).toCompletableFuture().get();
    }

    private static CompletionStage<List<String>> sendCommands(
            RedisAsyncCommands<String, String> commands, int numOfCommands) {
        commands.setAutoFlushCommands(false);
        List<CompletionStage<String>> responseFutures = new ArrayList<>(numOfCommands);
        for (int i = 0; i < numOfCommands; i++) {
            responseFutures.add(commands.get("key-" + i));
        }
        commands.flushCommands();
        commands.setAutoFlushCommands(true);
        return join(responseFutures);
    }

    private static long runLettuce(RedisAsyncCommands<String, String> commands, int numOfCommands)
            throws ExecutionException, InterruptedException {
        long startTime = System.nanoTime();
        CompletionStage<List<String>> stage = sendCommands(commands, numOfCommands);
        stage.toCompletableFuture().get();
        long endTime = System.nanoTime();
        return Duration.ofNanos(endTime - startTime).toMillis();
    }

    public static void main(String[] args) throws Throwable {
        String host = args[0];
        int port = Integer.valueOf(args[1]);
        int numOfCommands = Integer.valueOf(args[2]);

        RedisURI redisURI = RedisURI.Builder.redis(host, port).build();
        RedisClient client = RedisClient.create(redisURI);
        StatefulRedisConnection<String, String> connection = client.connect();
        RedisAsyncCommands<String, String> commands = connection.async();

        prepareData(commands, numOfCommands);

        long lettuceDuration = runLettuce(commands, numOfCommands);

        System.out.println(String.format("Number of queries: %d running time: %d ms",
                numOfCommands, lettuceDuration));

        connection.close();
        client.shutdown();
    }

}