ClientTest.java

package com.lambdaworks.redis;

import static com.google.code.tempusfugit.temporal.Duration.seconds;
import static com.google.code.tempusfugit.temporal.WaitFor.waitOrTimeout;
import static com.lambdaworks.redis.ScriptOutputType.STATUS;
import static org.assertj.core.api.Assertions.assertThat;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.ExpectedException;
import org.springframework.test.util.ReflectionTestUtils;

import com.google.code.tempusfugit.temporal.Condition;
import com.google.code.tempusfugit.temporal.Timeout;
import com.lambdaworks.redis.protocol.ConnectionWatchdog;
import com.lambdaworks.redis.server.RandomResponseServer;
import io.netty.channel.Channel;

public class ClientTest extends AbstractCommandTest {

    @Test
    public void testListenerAndCommandQueueingWhileReconnect() throws Exception {

        final TestConnectionListener listener = new TestConnectionListener();

        RedisClient client = new RedisClient(host, port);
        client.setOptions(new ClientOptions.Builder().autoReconnect(true).build());
        client.addListener(listener);

        assertThat(listener.onConnected).isNull();
        assertThat(listener.onDisconnected).isNull();
        assertThat(listener.onException).isNull();

        RedisAsyncConnectionImpl<String, String> connection = (RedisAsyncConnectionImpl) client.connectAsync();

        Channel channel = (Channel) ReflectionTestUtils.getField(connection.getChannelWriter(), "channel");
        ConnectionWatchdog connectionWatchdog = channel.pipeline().get(ConnectionWatchdog.class);

        // disable auto-reconnect for the test period until connection is killed and some commands are queued.
        connectionWatchdog.setListenOnChannelInactive(false);
        connectionWatchdog.setReconnectSuspended(true);

        connection.quit();
        Thread.sleep(100);
        
        // make sure, we received the connected and disconnected commands
        assertThat(listener.onConnected).isNotNull();
        assertThat(listener.onDisconnected).isNotNull();

        // reset listener state
        listener.onConnected = null;
        listener.onDisconnected = null;

        // queue some commands while we're down
        RedisFuture<String> set1 = connection.set("key1", value);
        RedisFuture<String> set2 = connection.set("key2", value);

        assertThat(set1.isDone()).isFalse();
        assertThat(set1.isCancelled()).isFalse();

        assertThat(connection.isOpen()).isFalse();
        
        // enable reconnect
        connectionWatchdog.setReconnectSuspended(false);
        connectionWatchdog.run(null);
        Thread.sleep(500);
        assertThat(connection.isOpen()).isTrue();
        
        
        // wait for command execution
        set1.get();
        set2.get();

        // check, that we got an event
        assertThat(listener.onConnected).isNotNull();
        assertThat(listener.onDisconnected).isNull();

        client.shutdown();
    }
}