, RExpirable {
* @return marked messages amount
*/
long ack(String groupName, StreamMessageId... ids);
-
+
/**
- * Returns pending messages by group name
+ * Returns common info about pending messages by group name.
*
* @param groupName - name of group
* @return result object
*/
+ PendingResult getPendingInfo(String groupName);
+
+ /*
+ * Use #getPendingInfo method
+ */
+ @Deprecated
PendingResult listPending(String groupName);
/**
- * Returns list of pending messages by group name.
+ * Returns list of common info about pending messages by group name.
* Limited by start Stream Message ID and end Stream Message ID and count.
*
* {@link StreamMessageId#MAX} is used as max Stream Message ID
* {@link StreamMessageId#MIN} is used as min Stream Message ID
*
+ * @see #pendingRangeAsync
+ *
* @param groupName - name of group
* @param startId - start Stream Message ID
* @param endId - end Stream Message ID
@@ -108,12 +116,14 @@ public interface RStream extends RStreamAsync, RExpirable {
List listPending(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
/**
- * Returns list of pending messages by group name and consumer name.
+ * Returns list of common info about pending messages by group and consumer name.
* Limited by start Stream Message ID and end Stream Message ID and count.
*
* {@link StreamMessageId#MAX} is used as max Stream Message ID
* {@link StreamMessageId#MIN} is used as min Stream Message ID
*
+ * @see #pendingRangeAsync
+ *
* @param consumerName - name of consumer
* @param groupName - name of group
* @param startId - start Stream Message ID
@@ -123,6 +133,41 @@ public interface RStream extends RStreamAsync, RExpirable {
*/
List listPending(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+ /**
+ * Returns stream data of pending messages by group name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Map> pendingRange(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
+
+ /**
+ * Returns stream data of pending messages by group and customer name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param consumerName - name of consumer
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Map> pendingRange(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+
/**
* Transfers ownership of pending messages by id to a new consumer
* by name if idle time of messages is greater than defined value.
@@ -655,7 +700,7 @@ public interface RStream extends RStreamAsync, RExpirable {
Map>> read(int count, long timeout, TimeUnit unit, StreamMessageId id, Map nameToId);
/**
- * Read stream data in range by specified start Stream Message ID (included) and end Stream Message ID (included).
+ * Returns stream data in range by specified start Stream Message ID (included) and end Stream Message ID (included).
*
* @param startId - start Stream Message ID
* @param endId - end Stream Message ID
@@ -664,7 +709,7 @@ public interface RStream extends RStreamAsync, RExpirable {
Map> range(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in range by specified start Stream Message ID (included) and end Stream Message ID (included).
+ * Returns stream data in range by specified start Stream Message ID (included) and end Stream Message ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream Message ID
@@ -674,7 +719,7 @@ public interface RStream extends RStreamAsync, RExpirable {
Map> range(int count, StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream Message ID (included) and end Stream Message ID (included).
+ * Returns stream data in reverse order in range by specified start Stream Message ID (included) and end Stream Message ID (included).
*
* @param startId - start Stream Message ID
* @param endId - end Stream Message ID
@@ -683,7 +728,7 @@ public interface RStream extends RStreamAsync, RExpirable {
Map> rangeReversed(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream Message ID (included) and end Stream Message ID (included).
+ * Returns stream data in reverse order in range by specified start Stream Message ID (included) and end Stream Message ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream Message ID
@@ -724,14 +769,14 @@ public interface RStream extends RStreamAsync, RExpirable {
StreamInfo getInfo();
/**
- * Returns list of objects with information about groups belonging to this stream.
+ * Returns list of common info about groups belonging to this stream.
*
* @return list of info objects
*/
List listGroups();
/**
- * Returns list of objects with information about group customers for specified groupName.
+ * Returns list of common info about group customers for specified groupName.
*
* @param groupName - name of group
* @return list of info objects
diff --git a/redisson/src/main/java/org/redisson/api/RStreamAsync.java b/redisson/src/main/java/org/redisson/api/RStreamAsync.java
index 188003b26..b190dfb4f 100644
--- a/redisson/src/main/java/org/redisson/api/RStreamAsync.java
+++ b/redisson/src/main/java/org/redisson/api/RStreamAsync.java
@@ -85,46 +85,91 @@ public interface RStreamAsync extends RExpirableAsync {
* @return marked messages amount
*/
RFuture ackAsync(String groupName, StreamMessageId... ids);
-
+
/**
- * Returns pending messages by group name
+ * Returns common info about pending messages by group name.
*
* @param groupName - name of group
* @return result object
*/
+ RFuture getPendingInfoAsync(String groupName);
+
+ /*
+ * Use #getPendingInfoAsync method
+ */
+ @Deprecated
RFuture listPendingAsync(String groupName);
/**
- * Returns list of pending messages by group name.
- * Limited by start stream id and end stream id and count.
+ * Returns list of common info about pending messages by group name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
*
- * {@link StreamMessageId#MAX} is used as max stream id
- * {@link StreamMessageId#MIN} is used as min stream id
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #pendingRangeAsync
*
* @param groupName - name of group
- * @param startId - start stream id
- * @param endId - end stream id
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
* @param count - amount of messages
* @return list
*/
RFuture> listPendingAsync(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
/**
- * Returns list of pending messages by group name and consumer name.
- * Limited by start stream id and end stream id and count.
+ * Returns list of common info about pending messages by group and consumer name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
*
- * {@link StreamMessageId#MAX} is used as max stream id
- * {@link StreamMessageId#MIN} is used as min stream id
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #pendingRangeAsync
*
* @param consumerName - name of consumer
* @param groupName - name of group
- * @param startId - start stream id
- * @param endId - end stream id
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
* @param count - amount of messages
* @return list
*/
RFuture> listPendingAsync(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+
+ /**
+ * Returns stream data of pending messages by group name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPendingAsync
+ *
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ RFuture>> pendingRangeAsync(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
+ /**
+ * Returns stream data of pending messages by group and customer name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPendingAsync
+ *
+ * @param consumerName - name of consumer
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ RFuture>> pendingRangeAsync(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+
/**
* Transfers ownership of pending messages by id to a new consumer
* by name if idle time of messages is greater than defined value.
@@ -645,7 +690,7 @@ public interface RStreamAsync extends RExpirableAsync {
RFuture>>> readAsync(int count, long timeout, TimeUnit unit, StreamMessageId id, Map nameToId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -654,7 +699,7 @@ public interface RStreamAsync extends RExpirableAsync {
RFuture>> rangeAsync(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
@@ -664,7 +709,7 @@ public interface RStreamAsync extends RExpirableAsync {
RFuture>> rangeAsync(int count, StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -673,7 +718,7 @@ public interface RStreamAsync extends RExpirableAsync {
RFuture>> rangeReversedAsync(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
diff --git a/redisson/src/main/java/org/redisson/api/RStreamReactive.java b/redisson/src/main/java/org/redisson/api/RStreamReactive.java
index 4f7cfd9a9..69480db39 100644
--- a/redisson/src/main/java/org/redisson/api/RStreamReactive.java
+++ b/redisson/src/main/java/org/redisson/api/RStreamReactive.java
@@ -89,11 +89,17 @@ public interface RStreamReactive extends RExpirableReactive {
Mono ack(String groupName, StreamMessageId... ids);
/**
- * Returns pending messages by group name
+ * Returns common info about pending messages by group name.
*
* @param groupName - name of group
* @return result object
*/
+ Mono getPendingInfo(String groupName);
+
+ /*
+ * Use #getPendingInfo method
+ */
+ @Deprecated
Mono listPending(String groupName);
/**
@@ -634,7 +640,7 @@ public interface RStreamReactive extends RExpirableReactive {
Mono>>> read(int count, long timeout, TimeUnit unit, StreamMessageId id, Map nameToId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -643,7 +649,7 @@ public interface RStreamReactive extends RExpirableReactive {
Mono>> range(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
@@ -653,7 +659,7 @@ public interface RStreamReactive extends RExpirableReactive {
Mono>> range(int count, StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -662,7 +668,7 @@ public interface RStreamReactive extends RExpirableReactive {
Mono>> rangeReversed(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
@@ -717,4 +723,39 @@ public interface RStreamReactive extends RExpirableReactive {
*/
Mono> listConsumers(String groupName);
+ /**
+ * Returns stream data of pending messages by group name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Mono>> pendingRange(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
+
+ /**
+ * Returns stream data of pending messages by group and customer name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param consumerName - name of consumer
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Mono>> pendingRange(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+
}
diff --git a/redisson/src/main/java/org/redisson/api/RStreamRx.java b/redisson/src/main/java/org/redisson/api/RStreamRx.java
index 9b70b9fc7..12f7dab7d 100644
--- a/redisson/src/main/java/org/redisson/api/RStreamRx.java
+++ b/redisson/src/main/java/org/redisson/api/RStreamRx.java
@@ -90,11 +90,17 @@ public interface RStreamRx extends RExpirableRx {
Single ack(String groupName, StreamMessageId... ids);
/**
- * Returns pending messages by group name
+ * Returns common info about pending messages by group name.
*
* @param groupName - name of group
* @return result object
*/
+ Single getPendingInfo(String groupName);
+
+ /*
+ * Use #getPendingInfo method
+ */
+ @Deprecated
Single listPending(String groupName);
/**
@@ -635,7 +641,7 @@ public interface RStreamRx extends RExpirableRx {
Single>>> read(int count, long timeout, TimeUnit unit, StreamMessageId id, Map nameToId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -644,7 +650,7 @@ public interface RStreamRx extends RExpirableRx {
Single>> range(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
@@ -654,7 +660,7 @@ public interface RStreamRx extends RExpirableRx {
Single>> range(int count, StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param startId - start Stream ID
* @param endId - end Stream ID
@@ -663,7 +669,7 @@ public interface RStreamRx extends RExpirableRx {
Single>> rangeReversed(StreamMessageId startId, StreamMessageId endId);
/**
- * Read stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
+ * Returns stream data in reverse order in range by specified start Stream ID (included) and end Stream ID (included).
*
* @param count - stream data size limit
* @param startId - start Stream ID
@@ -718,4 +724,39 @@ public interface RStreamRx extends RExpirableRx {
*/
Single> listConsumers(String groupName);
+ /**
+ * Returns stream data of pending messages by group name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Single>> pendingRange(String groupName, StreamMessageId startId, StreamMessageId endId, int count);
+
+ /**
+ * Returns stream data of pending messages by group and customer name.
+ * Limited by start Stream Message ID and end Stream Message ID and count.
+ *
+ * {@link StreamMessageId#MAX} is used as max Stream Message ID
+ * {@link StreamMessageId#MIN} is used as min Stream Message ID
+ *
+ * @see #listPending
+ *
+ * @param consumerName - name of consumer
+ * @param groupName - name of group
+ * @param startId - start Stream Message ID
+ * @param endId - end Stream Message ID
+ * @param count - amount of messages
+ * @return map
+ */
+ Single>> pendingRange(String groupName, String consumerName, StreamMessageId startId, StreamMessageId endId, int count);
+
}
diff --git a/redisson/src/main/java/org/redisson/client/RedisClient.java b/redisson/src/main/java/org/redisson/client/RedisClient.java
index 71935fe67..f3fa9707b 100644
--- a/redisson/src/main/java/org/redisson/client/RedisClient.java
+++ b/redisson/src/main/java/org/redisson/client/RedisClient.java
@@ -17,7 +17,6 @@ package org.redisson.client;
import java.net.InetAddress;
import java.net.InetSocketAddress;
-import java.net.URI;
import java.net.UnknownHostException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -28,6 +27,7 @@ import org.redisson.api.RFuture;
import org.redisson.client.handler.RedisChannelInitializer;
import org.redisson.client.handler.RedisChannelInitializer.Type;
import org.redisson.misc.RPromise;
+import org.redisson.misc.RedisURI;
import org.redisson.misc.RedissonPromise;
import io.netty.bootstrap.Bootstrap;
@@ -63,7 +63,7 @@ public final class RedisClient {
private final AtomicReference> resolvedAddrFuture = new AtomicReference>();
private final Bootstrap bootstrap;
private final Bootstrap pubSubBootstrap;
- private final URI uri;
+ private final RedisURI uri;
private InetSocketAddress resolvedAddr;
private final ChannelGroup channels;
diff --git a/redisson/src/main/java/org/redisson/client/RedisClientConfig.java b/redisson/src/main/java/org/redisson/client/RedisClientConfig.java
index cc6c3842f..50e260a18 100644
--- a/redisson/src/main/java/org/redisson/client/RedisClientConfig.java
+++ b/redisson/src/main/java/org/redisson/client/RedisClientConfig.java
@@ -20,7 +20,7 @@ import java.net.URI;
import java.util.concurrent.ExecutorService;
import org.redisson.config.SslProvider;
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.RedisURI;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.socket.SocketChannel;
@@ -35,7 +35,7 @@ import io.netty.util.Timer;
*/
public class RedisClientConfig {
- private URI address;
+ private RedisURI address;
private InetSocketAddress addr;
private Timer timer;
@@ -105,23 +105,23 @@ public class RedisClientConfig {
}
public RedisClientConfig setAddress(String host, int port) {
- this.address = URIBuilder.create("redis://" + host + ":" + port);
+ this.address = new RedisURI("redis://" + host + ":" + port);
return this;
}
public RedisClientConfig setAddress(String address) {
- this.address = URIBuilder.create(address);
+ this.address = new RedisURI(address);
return this;
}
- public RedisClientConfig setAddress(InetSocketAddress addr, URI address) {
+ public RedisClientConfig setAddress(InetSocketAddress addr, RedisURI address) {
this.addr = addr;
this.address = address;
return this;
}
- public RedisClientConfig setAddress(URI address) {
+ public RedisClientConfig setAddress(RedisURI address) {
this.address = address;
return this;
}
- public URI getAddress() {
+ public RedisURI getAddress() {
return address;
}
public InetSocketAddress getAddr() {
diff --git a/redisson/src/main/java/org/redisson/client/RedisRedirectException.java b/redisson/src/main/java/org/redisson/client/RedisRedirectException.java
index f5b80e979..f14fb54f9 100644
--- a/redisson/src/main/java/org/redisson/client/RedisRedirectException.java
+++ b/redisson/src/main/java/org/redisson/client/RedisRedirectException.java
@@ -15,9 +15,7 @@
*/
package org.redisson.client;
-import java.net.URI;
-
-import org.redisson.misc.URIBuilder;
+import org.redisson.misc.RedisURI;
/**
*
@@ -29,18 +27,18 @@ public class RedisRedirectException extends RedisException {
private static final long serialVersionUID = 181505625075250011L;
private final int slot;
- private final URI url;
+ private final RedisURI url;
public RedisRedirectException(int slot, String url) {
this.slot = slot;
- this.url = URIBuilder.create("redis://" + url);
+ this.url = new RedisURI("redis://" + url);
}
public int getSlot() {
return slot;
}
- public URI getUrl() {
+ public RedisURI getUrl() {
return url;
}
diff --git a/redisson/src/main/java/org/redisson/client/handler/ConnectionWatchdog.java b/redisson/src/main/java/org/redisson/client/handler/ConnectionWatchdog.java
index 3daaad16b..a1c8c77fc 100644
--- a/redisson/src/main/java/org/redisson/client/handler/ConnectionWatchdog.java
+++ b/redisson/src/main/java/org/redisson/client/handler/ConnectionWatchdog.java
@@ -177,14 +177,12 @@ public class ConnectionWatchdog extends ChannelInboundHandlerAdapter {
return;
}
- log.debug("blocking queue sent " + connection);
ChannelFuture future = connection.send(currentCommand);
- final CommandData cd = currentCommand;
future.addListener(new ChannelFutureListener() {
@Override
public void operationComplete(ChannelFuture future) throws Exception {
if (!future.isSuccess()) {
- log.error("Can't reconnect blocking queue to new connection. {}", cd);
+ log.error("Can't reconnect blocking queue by command: {} using connection: {}", currentCommand, connection);
}
}
});
diff --git a/redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java b/redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java
index e561ebdcc..beb672c62 100644
--- a/redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java
+++ b/redisson/src/main/java/org/redisson/cluster/ClusterConnectionManager.java
@@ -16,7 +16,6 @@
package org.redisson.cluster;
import java.net.InetSocketAddress;
-import java.net.URI;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.BitSet;
@@ -36,6 +35,7 @@ import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
import org.redisson.api.NodeType;
import org.redisson.api.RFuture;
@@ -58,8 +58,8 @@ import org.redisson.connection.MasterSlaveConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.SingleEntry;
import org.redisson.misc.RPromise;
+import org.redisson.misc.RedisURI;
import org.redisson.misc.RedissonPromise;
-import org.redisson.misc.URIBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -82,7 +82,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
private ScheduledFuture monitorFuture;
- private volatile URI lastClusterNode;
+ private volatile RedisURI lastClusterNode;
private RedisStrictCommand> clusterNodesCommand;
@@ -103,7 +103,8 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
Throwable lastException = null;
List failedMasters = new ArrayList();
- for (URI addr : cfg.getNodeAddresses()) {
+ for (String address : cfg.getNodeAddresses()) {
+ RedisURI addr = new RedisURI(address);
RFuture connectionFuture = connectToNode(cfg, addr, null, addr.getHost());
try {
RedisConnection connection = connectionFuture.syncUninterruptibly().getNow();
@@ -180,7 +181,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
@Override
- protected RedisClientConfig createRedisConfig(NodeType type, URI address, int timeout, int commandTimeout, String sslHostname) {
+ protected RedisClientConfig createRedisConfig(NodeType type, RedisURI address, int timeout, int commandTimeout, String sslHostname) {
RedisClientConfig result = super.createRedisConfig(type, address, timeout, commandTimeout, sslHostname);
result.setReadOnly(type == NodeType.SLAVE && config.getReadMode() != ReadMode.MASTER);
return result;
@@ -226,14 +227,15 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
MasterSlaveServersConfig config = create(cfg);
- config.setMasterAddress(partition.getMasterAddress());
+ config.setMasterAddress(partition.getMasterAddress().toString());
MasterSlaveEntry e;
List> futures = new ArrayList>();
if (config.checkSkipSlavesInit()) {
e = new SingleEntry(ClusterConnectionManager.this, config);
} else {
- config.setSlaveAddresses(partition.getSlaveAddresses());
+ Set slaveAddresses = partition.getSlaveAddresses().stream().map(r -> r.toString()).collect(Collectors.toSet());
+ config.setSlaveAddresses(slaveAddresses);
e = new MasterSlaveEntry(ClusterConnectionManager.this, config);
@@ -247,7 +249,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
}
- RFuture f = e.setupMasterEntry(config.getMasterAddress());
+ RFuture f = e.setupMasterEntry(new RedisURI(config.getMasterAddress()));
RPromise initFuture = new RedissonPromise();
futures.add(initFuture);
f.onComplete((res, ex3) -> {
@@ -275,12 +277,13 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
return result;
}
- private void scheduleClusterChangeCheck(ClusterServersConfig cfg, Iterator iterator) {
+ private void scheduleClusterChangeCheck(ClusterServersConfig cfg, Iterator iterator) {
monitorFuture = group.schedule(new Runnable() {
@Override
public void run() {
if (configEndpointHostName != null) {
- URI uri = cfg.getNodeAddresses().iterator().next();
+ String address = cfg.getNodeAddresses().iterator().next();
+ RedisURI uri = new RedisURI(address);
AddressResolver resolver = resolverGroup.getResolver(getGroup().next());
Future> allNodes = resolver.resolveAll(InetSocketAddress.createUnresolved(uri.getHost(), uri.getPort()));
allNodes.addListener(new FutureListener>() {
@@ -288,34 +291,34 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
public void operationComplete(Future> future) throws Exception {
AtomicReference lastException = new AtomicReference(future.cause());
if (!future.isSuccess()) {
- checkClusterState(cfg, Collections.emptyList().iterator(), lastException);
+ checkClusterState(cfg, Collections.emptyList().iterator(), lastException);
return;
}
- List nodes = new ArrayList();
+ List nodes = new ArrayList<>();
for (InetSocketAddress addr : future.getNow()) {
- URI node = URIBuilder.create(uri.getScheme() + "://" + addr.getAddress().getHostAddress() + ":" + addr.getPort());
- URI address = applyNatMap(node);
+ RedisURI node = new RedisURI(uri.getScheme() + "://" + addr.getAddress().getHostAddress() + ":" + addr.getPort());
+ RedisURI address = applyNatMap(node);
nodes.add(address);
}
- Iterator nodesIterator = nodes.iterator();
+ Iterator nodesIterator = nodes.iterator();
checkClusterState(cfg, nodesIterator, lastException);
}
});
} else {
AtomicReference lastException = new AtomicReference();
- Iterator nodesIterator = iterator;
+ Iterator nodesIterator = iterator;
if (nodesIterator == null) {
- List nodes = new ArrayList();
- List slaves = new ArrayList();
+ List nodes = new ArrayList<>();
+ List slaves = new ArrayList<>();
for (ClusterPartition partition : getLastPartitions()) {
if (!partition.isMasterFail()) {
nodes.add(partition.getMasterAddress());
}
- Set partitionSlaves = new HashSet(partition.getSlaveAddresses());
+ Set partitionSlaves = new HashSet<>(partition.getSlaveAddresses());
partitionSlaves.removeAll(partition.getFailedSlaveAddresses());
slaves.addAll(partitionSlaves);
}
@@ -332,7 +335,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}, cfg.getScanInterval(), TimeUnit.MILLISECONDS);
}
- private void checkClusterState(ClusterServersConfig cfg, Iterator iterator, AtomicReference lastException) {
+ private void checkClusterState(ClusterServersConfig cfg, Iterator iterator, AtomicReference lastException) {
if (!iterator.hasNext()) {
if (lastException.get() != null) {
log.error("Can't update cluster state", lastException.get());
@@ -343,7 +346,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
if (!getShutdownLatch().acquire()) {
return;
}
- URI uri = iterator.next();
+ RedisURI uri = iterator.next();
RFuture connectionFuture = connectToNode(cfg, uri, null, configEndpointHostName);
connectionFuture.onComplete((connection, e) -> {
if (e != null) {
@@ -358,7 +361,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
private void updateClusterState(ClusterServersConfig cfg, RedisConnection connection,
- Iterator iterator, URI uri, AtomicReference lastException) {
+ Iterator iterator, RedisURI uri, AtomicReference lastException) {
RFuture> future = connection.async(clusterNodesCommand);
future.onComplete((nodes, e) -> {
if (e != null) {
@@ -402,7 +405,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
MasterSlaveEntry entry = getEntry(currentPart.slots().nextSetBit(0));
// should be invoked first in order to remove stale failedSlaveAddresses
- Set addedSlaves = addRemoveSlaves(entry, currentPart, newPart);
+ Set addedSlaves = addRemoveSlaves(entry, currentPart, newPart);
// Do some slaves have changed state from failed to alive?
upDownSlaves(entry, currentPart, newPart, addedSlaves);
@@ -411,20 +414,20 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
}
- private void upDownSlaves(MasterSlaveEntry entry, ClusterPartition currentPart, ClusterPartition newPart, Set addedSlaves) {
- Set aliveSlaves = new HashSet(currentPart.getFailedSlaveAddresses());
+ private void upDownSlaves(MasterSlaveEntry entry, ClusterPartition currentPart, ClusterPartition newPart, Set addedSlaves) {
+ Set aliveSlaves = new HashSet<>(currentPart.getFailedSlaveAddresses());
aliveSlaves.removeAll(addedSlaves);
aliveSlaves.removeAll(newPart.getFailedSlaveAddresses());
- for (URI uri : aliveSlaves) {
+ for (RedisURI uri : aliveSlaves) {
currentPart.removeFailedSlaveAddress(uri);
if (entry.hasSlave(uri) && entry.slaveUp(uri, FreezeReason.MANAGER)) {
log.info("slave: {} has up for slot ranges: {}", uri, currentPart.getSlotRanges());
}
}
- Set failedSlaves = new HashSet(newPart.getFailedSlaveAddresses());
+ Set failedSlaves = new HashSet<>(newPart.getFailedSlaveAddresses());
failedSlaves.removeAll(currentPart.getFailedSlaveAddresses());
- for (URI uri : failedSlaves) {
+ for (RedisURI uri : failedSlaves) {
currentPart.addFailedSlaveAddress(uri);
if (entry.slaveDown(uri, FreezeReason.MANAGER)) {
log.warn("slave: {} has down for slot ranges: {}", uri, currentPart.getSlotRanges());
@@ -432,11 +435,11 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
}
- private Set addRemoveSlaves(MasterSlaveEntry entry, ClusterPartition currentPart, ClusterPartition newPart) {
- Set removedSlaves = new HashSet(currentPart.getSlaveAddresses());
+ private Set addRemoveSlaves(MasterSlaveEntry entry, ClusterPartition currentPart, ClusterPartition newPart) {
+ Set removedSlaves = new HashSet<>(currentPart.getSlaveAddresses());
removedSlaves.removeAll(newPart.getSlaveAddresses());
- for (URI uri : removedSlaves) {
+ for (RedisURI uri : removedSlaves) {
currentPart.removeSlaveAddress(uri);
if (entry.slaveDown(uri, FreezeReason.MANAGER)) {
@@ -444,9 +447,9 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
}
- Set addedSlaves = new HashSet(newPart.getSlaveAddresses());
+ Set addedSlaves = new HashSet<>(newPart.getSlaveAddresses());
addedSlaves.removeAll(currentPart.getSlaveAddresses());
- for (URI uri : addedSlaves) {
+ for (RedisURI uri : addedSlaves) {
RFuture future = entry.addSlave(uri);
future.onComplete((res, ex) -> {
if (ex != null) {
@@ -499,8 +502,8 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
ClusterPartition newMasterPart = find(newPartitions, slot);
// does partition has a new master?
if (!newMasterPart.getMasterAddress().equals(currentPart.getMasterAddress())) {
- URI newUri = newMasterPart.getMasterAddress();
- URI oldUri = currentPart.getMasterAddress();
+ RedisURI newUri = newMasterPart.getMasterAddress();
+ RedisURI oldUri = currentPart.getMasterAddress();
RFuture future = changeMaster(slot, newUri);
future.onComplete((res, e) -> {
@@ -693,10 +696,10 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
return result;
}
- public URI applyNatMap(URI address) {
+ public RedisURI applyNatMap(RedisURI address) {
String mappedAddress = natMap.get(address.getHost() + ":" + address.getPort());
if (mappedAddress != null) {
- return URIBuilder.create(address.getScheme() + "://" + mappedAddress);
+ return new RedisURI(address.getScheme() + "://" + mappedAddress);
}
return address;
}
@@ -718,7 +721,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
ClusterPartition partition = getPartition(partitions, id);
- URI address = applyNatMap(clusterNodeInfo.getAddress());
+ RedisURI address = applyNatMap(clusterNodeInfo.getAddress());
if (clusterNodeInfo.containsFlag(Flag.SLAVE)) {
slavePartition.setParent(partition);
@@ -751,10 +754,10 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
if (cp.getParent() != null && cp.getParent().getType() == Type.MASTER) {
ClusterPartition parent = cp.getParent();
- for (URI addr : cp.getSlaveAddresses()) {
+ for (RedisURI addr : cp.getSlaveAddresses()) {
parent.addSlaveAddress(addr);
}
- for (URI addr : cp.getFailedSlaveAddresses()) {
+ for (RedisURI addr : cp.getFailedSlaveAddresses()) {
parent.addFailedSlaveAddress(addr);
}
}
@@ -787,7 +790,7 @@ public class ClusterConnectionManager extends MasterSlaveConnectionManager {
}
@Override
- public URI getLastClusterNode() {
+ public RedisURI getLastClusterNode() {
return lastClusterNode;
}
diff --git a/redisson/src/main/java/org/redisson/cluster/ClusterNodeInfo.java b/redisson/src/main/java/org/redisson/cluster/ClusterNodeInfo.java
index 551c0c3fd..2b6facb83 100644
--- a/redisson/src/main/java/org/redisson/cluster/ClusterNodeInfo.java
+++ b/redisson/src/main/java/org/redisson/cluster/ClusterNodeInfo.java
@@ -15,11 +15,11 @@
*/
package org.redisson.cluster;
-import java.net.URI;
import java.util.EnumSet;
import java.util.HashSet;
import java.util.Set;
-import org.redisson.misc.URIBuilder;
+
+import org.redisson.misc.RedisURI;
/**
*
@@ -33,7 +33,7 @@ public class ClusterNodeInfo {
private final String nodeInfo;
private String nodeId;
- private URI address;
+ private RedisURI address;
private final Set flags = EnumSet.noneOf(Flag.class);
private String slaveOf;
@@ -50,11 +50,11 @@ public class ClusterNodeInfo {
this.nodeId = nodeId;
}
- public URI getAddress() {
+ public RedisURI getAddress() {
return address;
}
public void setAddress(String address) {
- this.address = URIBuilder.create(address);
+ this.address = new RedisURI(address);
}
public void addSlotRange(ClusterSlotRange range) {
diff --git a/redisson/src/main/java/org/redisson/cluster/ClusterPartition.java b/redisson/src/main/java/org/redisson/cluster/ClusterPartition.java
index 63abed343..12e78c547 100644
--- a/redisson/src/main/java/org/redisson/cluster/ClusterPartition.java
+++ b/redisson/src/main/java/org/redisson/cluster/ClusterPartition.java
@@ -15,12 +15,13 @@
*/
package org.redisson.cluster;
-import java.net.URI;
import java.util.BitSet;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
+import org.redisson.misc.RedisURI;
+
/**
*
* @author Nikita Koksharov
@@ -34,9 +35,9 @@ public class ClusterPartition {
private final String nodeId;
private boolean masterFail;
- private URI masterAddress;
- private final Set slaveAddresses = new HashSet();
- private final Set failedSlaves = new HashSet();
+ private RedisURI masterAddress;
+ private final Set slaveAddresses = new HashSet<>();
+ private final Set failedSlaves = new HashSet<>();
private final BitSet slots = new BitSet();
private final Set slotRanges = new HashSet();
@@ -123,30 +124,30 @@ public class ClusterPartition {
return slots.cardinality();
}
- public URI getMasterAddress() {
+ public RedisURI getMasterAddress() {
return masterAddress;
}
- public void setMasterAddress(URI masterAddress) {
+ public void setMasterAddress(RedisURI masterAddress) {
this.masterAddress = masterAddress;
}
- public void addFailedSlaveAddress(URI address) {
+ public void addFailedSlaveAddress(RedisURI address) {
failedSlaves.add(address);
}
- public Set getFailedSlaveAddresses() {
+ public Set getFailedSlaveAddresses() {
return Collections.unmodifiableSet(failedSlaves);
}
- public void removeFailedSlaveAddress(URI uri) {
+ public void removeFailedSlaveAddress(RedisURI uri) {
failedSlaves.remove(uri);
}
- public void addSlaveAddress(URI address) {
+ public void addSlaveAddress(RedisURI address) {
slaveAddresses.add(address);
}
- public Set getSlaveAddresses() {
+ public Set getSlaveAddresses() {
return Collections.unmodifiableSet(slaveAddresses);
}
- public void removeSlaveAddress(URI uri) {
+ public void removeSlaveAddress(RedisURI uri) {
slaveAddresses.remove(uri);
failedSlaves.remove(uri);
}
diff --git a/redisson/src/main/java/org/redisson/codec/SerializationCodec.java b/redisson/src/main/java/org/redisson/codec/SerializationCodec.java
index 66e0c677a..335dc327d 100644
--- a/redisson/src/main/java/org/redisson/codec/SerializationCodec.java
+++ b/redisson/src/main/java/org/redisson/codec/SerializationCodec.java
@@ -40,14 +40,22 @@ public class SerializationCodec extends BaseCodec {
@Override
public Object decode(ByteBuf buf, State state) throws IOException {
try {
- ByteBufInputStream in = new ByteBufInputStream(buf);
- ObjectInputStream inputStream;
- if (classLoader != null) {
- inputStream = new CustomObjectInputStream(classLoader, in);
- } else {
- inputStream = new ObjectInputStream(in);
+ //set thread context class loader to be the classLoader variable as there could be reflection
+ //done while reading from input stream which reflection will use thread class loader to load classes on demand
+ ClassLoader currentThreadClassLoader = Thread.currentThread().getContextClassLoader();
+ try {
+ ByteBufInputStream in = new ByteBufInputStream(buf);
+ ObjectInputStream inputStream;
+ if (classLoader != null) {
+ Thread.currentThread().setContextClassLoader(classLoader);
+ inputStream = new CustomObjectInputStream(classLoader, in);
+ } else {
+ inputStream = new ObjectInputStream(in);
+ }
+ return inputStream.readObject();
+ } finally {
+ Thread.currentThread().setContextClassLoader(currentThreadClassLoader);
}
- return inputStream.readObject();
} catch (IOException e) {
throw e;
} catch (Exception e) {
diff --git a/redisson/src/main/java/org/redisson/config/ClusterServersConfig.java b/redisson/src/main/java/org/redisson/config/ClusterServersConfig.java
index 9177499db..d786b58c6 100644
--- a/redisson/src/main/java/org/redisson/config/ClusterServersConfig.java
+++ b/redisson/src/main/java/org/redisson/config/ClusterServersConfig.java
@@ -15,15 +15,13 @@
*/
package org.redisson.config;
-import java.net.URI;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import org.redisson.misc.URIBuilder;
-
/**
*
* @author Nikita Koksharov
@@ -36,7 +34,7 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig nodeAddresses = new ArrayList();
+ private List nodeAddresses = new ArrayList<>();
/**
* Redis cluster scan interval in milliseconds
@@ -60,15 +58,13 @@ public class ClusterServersConfig extends BaseMasterSlaveServersConfig