import eu.rekawek.toxiproxy.model.Toxic;
import io.lettuce.core.ClientOptions;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.TimeoutOptions;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.api.async.RedisAsyncCommands;
import org.junit.After;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.ToxiproxyContainer;

import java.time.Duration;

import static org.hamcrest.Matchers.closeTo;
import static org.hamcrest.Matchers.lessThanOrEqualTo;
import static org.junit.Assert.assertThat;

public class DisconnectTest {
    @ClassRule
    public static Network network = Network.newNetwork();
    @ClassRule
    public static GenericContainer redisContainer = new GenericContainer<>("redis:5.0.5-alpine")
            .withExposedPorts(6379).withNetwork(network);
    @ClassRule
    public static ToxiproxyContainer toxiproxy = new ToxiproxyContainer().withNetwork(network);
    public static ToxiproxyContainer.ContainerProxy redisProxy;

    RedisClient redisClient;
    StatefulRedisConnection<String, String> connection;
    RedisAsyncCommands<String, String> redis;

    @BeforeClass
    public static void setUpClass() {
        redisProxy = toxiproxy.getProxy(redisContainer, 6379);
    }

    @Before
    public void connectToRedis() {
        redisClient = RedisClient.create(
                new RedisURI(redisProxy.getContainerIpAddress(), redisProxy.getProxyPort(), Duration.ofMillis(100)));
        redisClient.setOptions(ClientOptions.builder()
                .disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
                // enhance with setting socket read or write timeout?
                .timeoutOptions(TimeoutOptions.builder().connectionTimeout().build())
                .build());
        connection = redisClient.connect();
        redis = connection.async();
    }

    @After
    public void closeRedisClient() {
        connection.close();
        redisClient.shutdown();
    }

    @After
    public void resetProxy() throws Exception {
        for (Toxic t : redisProxy.toxics().getAll()) {
            t.remove();
        }
    }

    // Note: This test is using .get() on all the CompletableFutures to make the test
    // easier, but the principle should apply even if programming in a truly async style;
    // we would just expect to see the CompletionStages complete exceptionally immediately.
    @Test(timeout = 1000)
    public void commandsFailImmediatelyAfterTimeout() throws Exception {
        redis.set("test", "123").get();

        // Disconnect and see that command timeout works
        redisProxy.setConnectionCut(true);
        long start = System.nanoTime();
        redis.get("test").exceptionally(ignore -> null).toCompletableFuture().get();
        long stop = System.nanoTime();
        assertThat((stop - start) / 1000000., closeTo(100., 25.));

        // Want to detect that connection is not working anymore and further commands should
        // be rejected immediately
        start = System.nanoTime();
        redis.get("test").exceptionally(ignore -> null).toCompletableFuture().get();
        stop = System.nanoTime();
        assertThat((stop - start) / 1000000., lessThanOrEqualTo(1.));
    }
}