import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void insertionRemoval() throws NumberFormatException, InterruptedException, ExecutionException {
	RedisClient redis = RedisClient.create("redis://127.0.0.1:6379");
	StatefulRedisConnection<String, String> redisConn = redis.connect();
	RedisAsyncCommands<String, String> redisCmd = redisConn.async();
	
	long iterations = 1000;
	
	for (long i = 0; i < iterations; i++) {
		redisCmd.set(String.valueOf(i), String.valueOf(i + 1));
	}
	
	for (long i = 0; i < iterations; i++) {
		long v = Long.valueOf(redisCmd.get(String.valueOf(i)).get());
		assertEquals(i + 1, v);
	}
	
	for (long i = 0; i < iterations; i++) {
		redisCmd.del(String.valueOf(i));
	}
	
	redisConn.close();
	redis.shutdown();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@SuppressWarnings("unchecked")
protected AsyncNodeSelection<K, V> nodes(Predicate<RedisClusterNode> predicate, ClusterConnectionProvider.Intent intent,
        boolean dynamic) {

    NodeSelectionSupport<RedisAsyncCommands<K, V>, ?> selection;

    StatefulRedisClusterConnectionImpl<K, V> impl = (StatefulRedisClusterConnectionImpl<K, V>) connection;
    if (dynamic) {
        selection = new DynamicNodeSelection<>(impl.getClusterDistributionChannelWriter(), predicate, intent,
                StatefulRedisConnection::async);
    } else {
        selection = new StaticNodeSelection<>(impl.getClusterDistributionChannelWriter(), predicate, intent,
                StatefulRedisConnection::async);
    }

    NodeSelectionInvocationHandler h = new NodeSelectionInvocationHandler((AbstractNodeSelection<?, ?, ?, ?>) selection,
            RedisClusterAsyncCommands.class, ASYNC);
    return (AsyncNodeSelection<K, V>) Proxy.newProxyInstance(NodeSelectionSupport.class.getClassLoader(), new Class<?>[] {
            NodeSelectionAsyncCommands.class, AsyncNodeSelection.class }, h);
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test(timeout = 2000)
public void pipelinedMessage() throws Exception {
    pubsub.subscribe(channel);
    assertThat(channels.take()).isEqualTo(channel);
    RedisAsyncCommands<String, String> connection = client.connectAsync();

    connection.setAutoFlushCommands(false);
    connection.publish(channel, message);
    Thread.sleep(100);

    assertThat(channels).isEmpty();
    connection.flushCommands();

    assertThat(channels.take()).isEqualTo(channel);
    assertThat(messages.take()).isEqualTo(message);

    connection.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void wrappedConnectionShouldUseWrappers() throws Exception {

    GenericObjectPool<StatefulRedisConnection<String, String>> pool = ConnectionPoolSupport.createGenericObjectPool(
            () -> client.connect(), new GenericObjectPoolConfig());

    StatefulRedisConnection<String, String> connection = pool.borrowObject();
    RedisCommands<String, String> sync = connection.sync();

    assertThat(connection).isInstanceOf(StatefulRedisConnection.class).isNotInstanceOf(
            StatefulRedisClusterConnectionImpl.class);
    assertThat(Proxy.isProxyClass(connection.getClass())).isTrue();

    assertThat(sync).isInstanceOf(RedisCommands.class);
    assertThat(connection.async()).isInstanceOf(RedisAsyncCommands.class).isNotInstanceOf(RedisAsyncCommandsImpl.class);
    assertThat(connection.reactive()).isInstanceOf(RedisReactiveCommands.class).isNotInstanceOf(
            RedisReactiveCommandsImpl.class);
    assertThat(sync.getStatefulConnection()).isInstanceOf(StatefulRedisConnection.class)
            .isNotInstanceOf(StatefulRedisConnectionImpl.class).isSameAs(connection);

    sync.close();
    pool.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void wrappedMasterSlaveConnectionShouldUseWrappers() throws Exception {

    GenericObjectPool<StatefulRedisMasterSlaveConnection<String, String>> pool = ConnectionPoolSupport
            .createGenericObjectPool(() -> MasterSlave.connect(client, new StringCodec(), RedisURI.create(host, port)),
                    new GenericObjectPoolConfig());

    StatefulRedisMasterSlaveConnection<String, String> connection = pool.borrowObject();
    RedisCommands<String, String> sync = connection.sync();

    assertThat(connection).isInstanceOf(StatefulRedisMasterSlaveConnection.class);
    assertThat(Proxy.isProxyClass(connection.getClass())).isTrue();

    assertThat(sync).isInstanceOf(RedisCommands.class);
    assertThat(connection.async()).isInstanceOf(RedisAsyncCommands.class).isNotInstanceOf(RedisAsyncCommandsImpl.class);
    assertThat(connection.reactive()).isInstanceOf(RedisReactiveCommands.class).isNotInstanceOf(
            RedisReactiveCommandsImpl.class);
    assertThat(sync.getStatefulConnection()).isInstanceOf(StatefulRedisConnection.class)
            .isNotInstanceOf(StatefulRedisConnectionImpl.class).isSameAs(connection);

    sync.close();
    pool.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void plainConnectionShouldNotUseWrappers() throws Exception {

    GenericObjectPool<StatefulRedisConnection<String, String>> pool = ConnectionPoolSupport.createGenericObjectPool(
            () -> client.connect(), new GenericObjectPoolConfig(), false);

    StatefulRedisConnection<String, String> connection = pool.borrowObject();
    RedisCommands<String, String> sync = connection.sync();

    assertThat(connection).isInstanceOf(StatefulRedisConnection.class).isNotInstanceOf(
            StatefulRedisClusterConnectionImpl.class);
    assertThat(Proxy.isProxyClass(connection.getClass())).isFalse();

    assertThat(sync).isInstanceOf(RedisCommands.class);
    assertThat(connection.async()).isInstanceOf(RedisAsyncCommands.class).isInstanceOf(RedisAsyncCommandsImpl.class);
    assertThat(connection.reactive()).isInstanceOf(RedisReactiveCommands.class).isInstanceOf(
            RedisReactiveCommandsImpl.class);
    assertThat(sync.getStatefulConnection()).isInstanceOf(StatefulRedisConnection.class).isInstanceOf(
            StatefulRedisConnectionImpl.class);

    pool.returnObject(connection);
    pool.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void requestQueueSize() {

    client.setOptions(ClientOptions.builder().requestQueueSize(10).build());

    RedisAsyncCommands<String, String> connection = client.connect().async();
    getConnectionWatchdog(connection.getStatefulConnection()).setListenOnChannelInactive(false);

    connection.quit();

    Wait.untilTrue(() -> !connection.isOpen()).waitOrTimeout();

    for (int i = 0; i < 10; i++) {
        connection.ping();
    }

    try {
        connection.ping();
        fail("missing RedisException");
    } catch (RedisException e) {
        assertThat(e).hasMessageContaining("Request queue size exceeded");
    }

    connection.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void disconnectedWithoutReconnect() {

    client.setOptions(ClientOptions.builder().autoReconnect(false).build());

    RedisAsyncCommands<String, String> connection = client.connect().async();

    connection.quit();
    Wait.untilTrue(() -> !connection.isOpen()).waitOrTimeout();
    try {
        connection.get(key);
    } catch (Exception e) {
        assertThat(e).isInstanceOf(RedisException.class).hasMessageContaining("not connected");
    } finally {
        connection.close();
    }
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void disconnectedRejectCommands() {

    client.setOptions(ClientOptions.builder().disconnectedBehavior(ClientOptions.DisconnectedBehavior.REJECT_COMMANDS)
            .build());

    RedisAsyncCommands<String, String> connection = client.connect().async();

    getConnectionWatchdog(connection.getStatefulConnection()).setListenOnChannelInactive(false);
    connection.quit();
    Wait.untilTrue(() -> !connection.isOpen()).waitOrTimeout();
    try {
        connection.get(key);
    } catch (Exception e) {
        assertThat(e).isInstanceOf(RedisException.class).hasMessageContaining("not connected");
    } finally {
        connection.close();
    }
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void asyncPoolPerformanceTest() throws Exception {

    RedisConnectionPool<RedisAsyncCommands<String, String>> pool = client.asyncPool();
    RedisAsyncConnection<String, String> c1 = pool.allocateConnection();

    c1.ping();
    Stopwatch stopwatch = Stopwatch.createStarted();

    for (int i = 0; i < 1000; i++) {
        c1.ping();
    }

    long elapsed = stopwatch.stop().elapsed(TimeUnit.MILLISECONDS);

    log.info("asyncPoolPerformanceTest Duration: " + elapsed + "ms");

    c1.close();
    pool.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
private void testUnwrap(AbstractRedisClient client) {
    Assert.assertTrue(cache.unwrap(AbstractRedisClient.class) instanceof AbstractRedisClient);
    if (client instanceof RedisClient) {
        Assert.assertTrue(cache.unwrap(RedisClient.class) instanceof RedisClient);
        Assert.assertTrue(cache.unwrap(RedisCommands.class) instanceof RedisCommands);
        Assert.assertTrue(cache.unwrap(RedisAsyncCommands.class) instanceof RedisAsyncCommands);
        Assert.assertTrue(cache.unwrap(RedisReactiveCommands.class) instanceof RedisReactiveCommands);
    } else {
        Assert.assertTrue(cache.unwrap(RedisClusterClient.class) instanceof RedisClusterClient);
        Assert.assertTrue(cache.unwrap(RedisClusterCommands.class) instanceof RedisClusterCommands);
        Assert.assertTrue(cache.unwrap(RedisClusterAsyncCommands.class) instanceof RedisClusterAsyncCommands);
        Assert.assertTrue(cache.unwrap(RedisClusterReactiveCommands.class) instanceof RedisClusterReactiveCommands);
    }
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void tests() throws Exception {
    if (RedisLettuceCacheTest.checkOS()) {
        System.setProperty("spring.profiles.active", "redislettuce4-cluster");
    } else {
        System.setProperty("spring.profiles.active", "redislettuce4");
    }
    context = SpringApplication.run(RedisLettuce4StarterTest.class);
    doTest();
    A bean = context.getBean(A.class);
    bean.test();

    RedisClient t1 = (RedisClient) context.getBean("defaultClient");
    RedisClient t2 = (RedisClient) context.getBean("a1Client");
    Assert.assertNotNull(t1);
    Assert.assertNotNull(t2);
    Assert.assertNotSame(t1, t2);

    AutoConfigureBeans acb = context.getBean(AutoConfigureBeans.class);

    String key = "remote.A1";
    Assert.assertTrue(new Lettuce4Factory(acb, key, StatefulRedisConnection.class).getObject() instanceof StatefulRedisConnection);
    Assert.assertTrue(new Lettuce4Factory(acb, key, RedisCommands.class).getObject() instanceof RedisCommands);
    Assert.assertTrue(new Lettuce4Factory(acb, key, RedisAsyncCommands.class).getObject() instanceof RedisAsyncCommands);
    Assert.assertTrue(new Lettuce4Factory(acb, key, RedisReactiveCommands.class).getObject() instanceof RedisReactiveCommands);

    if (RedisLettuceCacheTest.checkOS()) {
        key = "remote.A2";
        Assert.assertTrue(new Lettuce4Factory(acb, key , RedisClusterClient.class).getObject() instanceof RedisClusterClient);
        Assert.assertTrue(new Lettuce4Factory(acb, key , RedisClusterCommands.class).getObject() instanceof RedisClusterCommands);
        Assert.assertTrue(new Lettuce4Factory(acb, key , RedisClusterAsyncCommands.class).getObject() instanceof RedisClusterAsyncCommands);
        Assert.assertTrue(new Lettuce4Factory(acb, key , RedisClusterReactiveCommands.class).getObject() instanceof RedisClusterReactiveCommands);
    }
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
public <R> R map(Function<? super RedisAsyncCommands<String, String>, ? extends R> mapper) {
	RedisAsyncCommands<String, String> conn = pool.allocateConnection();
	try {
		return mapper.apply(conn);
	} finally {
		pool.freeConnection(conn);
	}
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
public void with(Consumer<? super RedisAsyncCommands<String, String>> action) {
	RedisAsyncCommands<String, String> conn = pool.allocateConnection();
	try {
		action.accept(conn);
	} finally {
		pool.freeConnection(conn);
	}
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
/**
 * Creates a connection pool for asynchronous connections. Please keep in mind to free all collections and close the pool
 * once you do not need it anymore. Requires Apache commons-pool2 dependency.
 *
 * @param codec Use this codec to encode/decode keys and values, must not be {@literal null}
 * @param maxIdle max idle connections in pool
 * @param maxActive max active connections in pool
 * @param <K> Key type
 * @param <V> Value type
 * @return a new {@link RedisConnectionPool} instance
 * @deprecated Will be removed in future versions. Use {@link ConnectionPoolSupport}.
 */
@Deprecated
public <K, V> RedisConnectionPool<RedisAsyncCommands<K, V>> asyncPool(final RedisCodec<K, V> codec, int maxIdle,
        int maxActive) {

    checkPoolDependency();
    checkForRedisURI();
    LettuceAssert.notNull(codec, "RedisCodec must not be null");

    long maxWait = makeTimeout();
    RedisConnectionPool<RedisAsyncCommands<K, V>> pool = new RedisConnectionPool<>(
            new RedisConnectionPool.RedisConnectionProvider<RedisAsyncCommands<K, V>>() {
                @Override
                public RedisAsyncCommands<K, V> createConnection() {
                    return connectStandalone(codec, redisURI, defaultTimeout()).async();
                }

                @Override
                @SuppressWarnings({ "rawtypes", "unchecked" })
                public Class<? extends RedisAsyncCommands<K, V>> getComponentType() {
                    return (Class) RedisAsyncCommands.class;
                }
            }, maxActive, maxIdle, maxWait);

    pool.addListener(closeableResources::remove);

    closeableResources.add(pool);

    return pool;
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
protected List<RedisFuture<?>> triggerSet(RedisAsyncCommands<String, String> connection, int iterations) {
    List<RedisFuture<?>> futures = new ArrayList<>();
    for (int i = 0; i < iterations; i++) {
        futures.add(connection.set(key(i), value(i)));
    }
    return futures;
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
@Test
public void disconnectedAcceptCommands() {

    client.setOptions(ClientOptions.builder().autoReconnect(false)
            .disconnectedBehavior(ClientOptions.DisconnectedBehavior.ACCEPT_COMMANDS).build());

    RedisAsyncCommands<String, String> connection = client.connect().async();

    connection.quit();
    Wait.untilTrue(() -> !connection.isOpen()).waitOrTimeout();
    connection.get(key);
    connection.close();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
private void suspendConnection(RedisClusterAsyncCommands<String, String> asyncCommands) {
    Connections.getConnectionWatchdog(((RedisAsyncCommands<?, ?>) asyncCommands).getStatefulConnection())
            .setReconnectSuspended(true);
    asyncCommands.quit();

    Wait.untilTrue(() -> !asyncCommands.isOpen()).waitOrTimeout();
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
public ClusterRule(RedisClusterClient clusterClient, int... ports) {
    this.clusterClient = clusterClient;
    this.ports = ports;

    for (int port : ports) {
        RedisAsyncCommands<String, String> connection = clusterClient.connectToNode(
                new InetSocketAddress("localhost", port)).async();
        connectionCache.put(port, connection);
    }
}
 

import com.lambdaworks.redis.api.async.RedisAsyncCommands; //导入依赖的package包/类
/**
 *
 * @return true if the cluster state is {@code ok} and there are no failing nodes
 */
public boolean isStable() {

    for (RedisAsyncCommands<String, String> commands : connectionCache.values()) {
        try {
            RedisCommands<String, String> sync = commands.getStatefulConnection().sync();
            String info = sync.clusterInfo();
            if (info != null && info.contains("cluster_state:ok")) {

                String s = sync.clusterNodes();
                Partitions parse = ClusterPartitionParser.parse(s);

                for (RedisClusterNode redisClusterNode : parse) {
                    if (redisClusterNode.getFlags().contains(RedisClusterNode.NodeFlag.FAIL)
                            || redisClusterNode.getFlags().contains(RedisClusterNode.NodeFlag.EVENTUAL_FAIL)
                            || redisClusterNode.getFlags().contains(RedisClusterNode.NodeFlag.HANDSHAKE)) {
                        return false;
                    }
                }

            } else {
                return false;
            }
        } catch (Exception e) {
            // nothing to do
        }
    }

    return true;
}