import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
/**
 * Connects to the Redis server.
 * @param configuration The configuration object.
 * @throws Exception If an error occurs.
 */
public void connect(Configuration configuration) throws Exception {
    if(sync != null) {
        return;
    }
    RedisURI.Builder uri = RedisURI.Builder.redis(configuration.getRedisHost(), configuration.getRedisPort())
            .withDatabase(configuration.getRedisIndex());
    if(!configuration.getRedisAuth().isEmpty()) {
        uri.withPassword(configuration.getRedisAuth());
    }
    RedisClient client = RedisClient.create(uri.build());
    StatefulRedisConnection<String, String> connection = client.connect();
    sync = connection.sync();
    for(Category category : Category.values()) {
        logger.info("Registered the category {} with the type {}.", category, category.getEntry().getType());
        category.getEntry().setCategory(category);
    }
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的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.StatefulRedisConnection; //导入依赖的package包/类
public void ping(final PingCallback callback) {
    Consumer<StatefulRedisConnection> connectionConsumer = new Consumer<StatefulRedisConnection>() {
        @Override
        public void accept(StatefulRedisConnection statefulRedisConnection) {
            final CompletableFuture<String> future = statefulRedisConnection.async().ping().toCompletableFuture();

            future.whenCompleteAsync((pong, th) -> {
                if(th != null){
                    callback.fail(th);
                }else{
                    callback.pong(pong);
                }
            }, executors);
        }
    };
    Consumer<Throwable> throwableConsumer = new Consumer<Throwable>() {
        @Override
        public void accept(Throwable e) {
            callback.fail(e);
            log.error("[ping]" + hostPort, e);
        }
    };
    asyncExecute(connectionConsumer, throwableConsumer);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
public void role(RollCallback callback) {
    Consumer<StatefulRedisConnection> connectionConsumer = new Consumer<StatefulRedisConnection>() {
        @Override
        public void accept(StatefulRedisConnection statefulRedisConnection) {
            final CompletableFuture<List<Object>> future = statefulRedisConnection.async().role().toCompletableFuture();

            future.whenCompleteAsync((role, th) -> {
                if (th != null) {
                    callback.fail(th);
                } else {
                    callback.role((String) role.get(0));
                }
            }, executors);
        }
    };
    asyncExecute(connectionConsumer, null);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
public void serverInfo(Callbackable<String> callback) {
    String serverInfoSection = "server";
    Consumer<StatefulRedisConnection> connectionConsumer = (connection) -> {
        CompletableFuture<String> future = connection.async().info(serverInfoSection).toCompletableFuture();
        future.whenCompleteAsync((info, th) -> {
            if(th != null){
                log.error("[info]{}", hostPort, th);
                callback.fail(th);
            }else{
                callback.success(info);
            }
        }, executors);
    };

    Consumer<Throwable> throwableConsumer = (throwable) -> {
        callback.fail(throwable);
        log.error("[info]{}", hostPort, throwable);
    };

    asyncExecute(connectionConsumer, throwableConsumer);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
public void conf(String confSection, Callbackable<List<String>> callback) {
    Consumer<StatefulRedisConnection> connectionConsumer = (connection) -> {
        CompletableFuture<List<String>> future = connection.async().configGet(confSection).toCompletableFuture();
        future.whenCompleteAsync((conf, throwable) -> {
            if(throwable != null) {
                log.error("[conf]Executing conf command error", throwable);
                callback.fail(throwable);
            } else {
                callback.success(conf);
            }
        });
    };

    Consumer<Throwable> throwableConsumer = (throwable) -> {
        callback.fail(throwable);
        log.error("[conf]{}", hostPort, throwable);
    };

    asyncExecute(connectionConsumer, throwableConsumer);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
@SuppressWarnings("unused")
private void publish() {

	scheduled.scheduleAtFixedRate(new AbstractExceptionLogTask() {

		StatefulRedisConnection<String, String> redisConnection;
		private RedisClient redisClient;
		{
			redisClient = RedisClient.create(clientResources, redisURI);
		}

		@Override
		public void doRun() {

			if (redisConnection == null) {
				redisConnection = redisClient.connect();
			}

			logger.info("[run][publish]{}", channel);
			redisConnection.async().publish(channel, randomString(10));

		}
	}, 0, 5, TimeUnit.SECONDS);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
protected StatefulRedisConnection<String, String> getConnection()
{
    if (connection == null) {
        synchronized (syncObject)
        {
            if (connection == null)
            {
                Logging.write("$");
                RedisClient client = getRedisClient();
                connection = client.connect();
                TrySetClientName();
            }
        }
    }

    return connection;
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
@Override
@SuppressWarnings("unchecked")
protected Object handleInvocation(Object proxy, Method method, Object[] args) throws Throwable {

    try {

        Method targetMethod = methodCache.get(method);
        Object result = targetMethod.invoke(asyncApi, args);

        if (result instanceof RedisFuture) {
            RedisFuture<?> command = (RedisFuture<?>) result;
            if (!method.getName().equals("exec") && !method.getName().equals("multi")) {
                if (connection instanceof StatefulRedisConnection && ((StatefulRedisConnection) connection).isMulti()) {
                    return null;
                }
            }

            LettuceFutures.awaitOrCancel(command, connection.getTimeout(), connection.getTimeoutUnit());
            return command.get();
        }
        return result;
    } catch (InvocationTargetException e) {
        throw e.getTargetException();
    }

}
 

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

    try (StatefulRedisConnection<String, String> connect = client.connect()) {

        Wait.untilEquals(false, connect::isOpen).waitOrTimeout();

        connect.sync().ping();
    } catch (RedisException e) {
        if (e.getCause() instanceof IOException) {
            assertThat(e).hasCauseInstanceOf(IOException.class);
        } else {
            assertThat(e.getCause()).hasMessageContaining("DENIED");
        }
    }
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
private <K, V> ConnectionFuture<StatefulRedisConnection<K, V>> connectStandaloneAsync(RedisCodec<K, V> codec,
        RedisURI redisURI, Timeout timeout) {

    assertNotNull(codec);
    checkValidRedisURI(redisURI);

    logger.debug("Trying to get a Redis connection for: " + redisURI);

    CommandHandler<K, V> handler = new CommandHandler<>(clientOptions, clientResources);

    StatefulRedisConnectionImpl<K, V> connection = newStatefulRedisConnection(handler, codec, timeout.timeout,
            timeout.timeUnit);
    ConnectionFuture<StatefulRedisConnection<K, V>> future = connectStatefulAsync(handler, connection, redisURI);

    future.whenComplete((channelHandler, throwable) -> {

        if (throwable != null) {
            connection.close();
        }
    });

    return future;
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
/**
 * Close stale connections.
 */
public void closeStaleConnections() {
    logger.debug("closeStaleConnections() count before expiring: {}", getConnectionCount());

    Set<ConnectionKey> stale = getStaleConnectionKeys();

    for (ConnectionKey connectionKey : stale) {
        StatefulRedisConnection<K, V> connection = connections.get(connectionKey);
        if (connection != null) {
            connections.remove(connectionKey);
            connection.close();
        }
    }

    logger.debug("closeStaleConnections() count after expiring: {}", getConnectionCount());
}
 

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

    SoftReferenceObjectPool<StatefulRedisConnection<String, String>> pool = ConnectionPoolSupport
            .createSoftReferenceObjectPool(() -> client.connect());

    StatefulRedisConnection<String, String> usedConnection = null;
    try (StatefulRedisConnection<String, String> connection = pool.borrowObject()) {

        RedisCommands<String, String> sync = connection.sync();
        sync.ping();

        usedConnection = connection;
    }

    try {
        usedConnection.isMulti();
        fail("Missing RedisException");
    } catch (RedisException e) {
        assertThat(e).hasMessageContaining("deallocated");
    }

    pool.close();
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
@Override
public void call(Subscriber<? super T> subscriber) {

    // Reuse the first command but then discard it.
    RedisCommand<K, V, T> command = this.command;
    if (command == null) {
        command = commandSupplier.get();
    }

    if (command.getOutput() instanceof StreamingOutput<?>) {
        StreamingOutput<T> streamingOutput = (StreamingOutput<T>) command.getOutput();

        if (connection instanceof StatefulRedisConnection<?, ?> && ((StatefulRedisConnection) connection).isMulti()) {
            streamingOutput.setSubscriber(new DelegatingWrapper<>(new ObservableSubscriberWrapper<>(subscriber),
                    streamingOutput.getSubscriber()));
        } else {
            streamingOutput.setSubscriber(new ObservableSubscriberWrapper<>(subscriber));
        }
    }

    connection.dispatch(new ObservableCommand<>(command, subscriber, dissolve));

    this.command = null;

}
 

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

    MasterSlaveTopologyRefresh refresh = new MasterSlaveTopologyRefresh(connectionFactory, provider);

    CompletableFuture<StatefulRedisConnection<String, String>> master = CompletableFuture.completedFuture(connection);
    CompletableFuture<StatefulRedisConnection<String, String>> slave = CompletableFuture.completedFuture(connection);
    when(connectionFactory.connectToNodeAsync(any(), any())).thenReturn((CompletableFuture) master,
            (CompletableFuture) slave);

    RedisURI redisURI = new RedisURI();
    redisURI.setTimeout(1);
    redisURI.setUnit(TimeUnit.MILLISECONDS);

    List<RedisNodeDescription> nodes = refresh.getNodes(redisURI);

    assertThat(nodes).hasSize(2);
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的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.StatefulRedisConnection; //导入依赖的package包/类
@Test
public void operateOnNodeConnection() {

    sync.set(KEY_A, value);
    sync.set(KEY_B, "d");

    StatefulRedisConnection<String, String> statefulRedisConnection = sync.getStatefulConnection().getConnection(
            TestSettings.hostAddr(), port2);

    RedisClusterCommands<String, String> connection = statefulRedisConnection.sync();

    assertThat(connection.get(KEY_A)).isEqualTo(value);
    try {
        connection.get(KEY_B);
        fail("missing RedisCommandExecutionException: MOVED");
    } catch (RedisException e) {
        assertThat(e).hasMessageContaining("MOVED");
    }
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
@Test
@SuppressWarnings("unchecked")
public void authInvalidPassword() throws Exception {
    RedisAsyncConnection<String, String> async = client.connectAsync();
    try {
        async.auth("invalid");
        fail("Authenticated with invalid password");
    } catch (RedisException e) {
        assertThat(e.getMessage()).isEqualTo("ERR Client sent AUTH, but no password is set");
        StatefulRedisConnection<String, String> statefulRedisConnection = (StatefulRedisConnection<String, String>) ReflectionTestUtils
                .getField(async, "connection");
        assertThat(ReflectionTestUtils.getField(statefulRedisConnection, "password")).isNull();
    } finally {
        async.close();
    }
}
 

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

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

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

    try {
        connection.isMulti();
        fail("Missing RedisException");
    } catch (RedisException e) {
        assertThat(e).hasMessageContaining("deallocated");
    }

    pool.close();
}
 

import com.lambdaworks.redis.api.StatefulRedisConnection; //导入依赖的package包/类
private StatefulRedisConnection<K, V>[] getConnections(
        CompletableFuture<StatefulRedisConnection<K, V>>[] selectedReaderCandidates) {

    List<StatefulRedisConnection<K, V>> connections = new ArrayList<>(selectedReaderCandidates.length);

    for (CompletableFuture<StatefulRedisConnection<K, V>> candidate : selectedReaderCandidates) {

        try {
            connections.add(candidate.join());
        } catch (Exception o_O) {
        }
    }

    StatefulRedisConnection<K, V>[] result = new StatefulRedisConnection[connections.size()];
    connections.toArray(result);
    return result;
}