github-forks
/
redisson
mirror of https://github.com/redisson/redisson.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

SubscribesConnectionEntry renamed

Browse Source
pull/297/head
Nikita 9 years ago
parent 1f4cb96ac5
commit 556b0b2a79
18 changed files with 73 additions and 73 deletions
Show Stats Download Patch File Download Diff File

2
src/main/java/org/redisson/cluster/ClusterConnectionListener.java
Unescape Escape View File

@ -20,7 +20,7 @@ import org.redisson.client.RedisException;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.connection.DefaultConnectionListener;
import org.redisson.connection.FutureConnectionListener;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
public class ClusterConnectionListener extends DefaultConnectionListener {

2
src/main/java/org/redisson/cluster/ClusterConnectionManager.java
Unescape Escape View File

@ -36,7 +36,7 @@ import org.redisson.cluster.ClusterNodeInfo.Flag;
import org.redisson.connection.CRC16;
import org.redisson.connection.MasterSlaveConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

14
src/main/java/org/redisson/connection/SubscribesConnectionEntry.java → src/main/java/org/redisson/connection/ClientConnectionsEntry.java
Unescape Escape View File

@ -31,7 +31,7 @@ import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.FutureListener;
import io.netty.util.concurrent.Promise;
public class SubscribesConnectionEntry {
public class ClientConnectionsEntry {
final Logger log = LoggerFactory.getLogger(getClass());
@ -48,17 +48,17 @@ public class SubscribesConnectionEntry {
public enum NodeType {SLAVE, MASTER}
private final NodeType nodeType;
private final ConnectionListener connectListener;
private final ConnectionListener connectionListener;
private final Queue<RedisConnection> freeConnections = new ConcurrentLinkedQueue<RedisConnection>();
private final AtomicInteger freeConnectionsCounter = new AtomicInteger();
private final AtomicInteger failedAttempts = new AtomicInteger();
public SubscribesConnectionEntry(RedisClient client, int poolSize, int subscribePoolSize, ConnectionListener connectListener, NodeType serverMode) {
public ClientConnectionsEntry(RedisClient client, int poolSize, int subscribePoolSize, ConnectionListener connectionListener, NodeType serverMode) {
this.client = client;
this.freeConnectionsCounter.set(poolSize);
this.connectListener = connectListener;
this.connectionListener = connectionListener;
this.nodeType = serverMode;
freeSubscribeConnectionsCounter.set(subscribePoolSize);
}
@ -145,7 +145,7 @@ public class SubscribesConnectionEntry {
log.debug("new connection created: {}", conn);
FutureConnectionListener<RedisConnection> listener = new FutureConnectionListener<RedisConnection>(connectionFuture, conn);
connectListener.onConnect(config, nodeType, listener);
connectionListener.onConnect(config, nodeType, listener);
listener.executeCommands();
addReconnectListener(config, conn);
@ -160,7 +160,7 @@ public class SubscribesConnectionEntry {
@Override
public void onReconnect(RedisConnection conn, Promise<RedisConnection> connectionFuture) {
FutureConnectionListener<RedisConnection> listener = new FutureConnectionListener<RedisConnection>(connectionFuture, conn);
connectListener.onConnect(config, nodeType, listener);
connectionListener.onConnect(config, nodeType, listener);
listener.executeCommands();
}
});
@ -180,7 +180,7 @@ public class SubscribesConnectionEntry {
log.debug("new pubsub connection created: {}", conn);
FutureConnectionListener<RedisPubSubConnection> listener = new FutureConnectionListener<RedisPubSubConnection>(connectionFuture, conn);
connectListener.onConnect(config, nodeType, listener);
connectionListener.onConnect(config, nodeType, listener);
listener.executeCommands();
addReconnectListener(config, conn);

2
src/main/java/org/redisson/connection/ConnectionListener.java
Unescape Escape View File

@ -17,7 +17,7 @@ package org.redisson.connection;
import org.redisson.MasterSlaveServersConfig;
import org.redisson.client.RedisException;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
public interface ConnectionListener {

2
src/main/java/org/redisson/connection/ConnectionManager.java
Unescape Escape View File

@ -28,7 +28,7 @@ import org.redisson.client.RedisPubSubListener;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.cluster.ClusterSlotRange;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.InfinitySemaphoreLatch;
import io.netty.channel.EventLoopGroup;

2
src/main/java/org/redisson/connection/DefaultConnectionListener.java
Unescape Escape View File

@ -18,7 +18,7 @@ package org.redisson.connection;
import org.redisson.MasterSlaveServersConfig;
import org.redisson.client.RedisException;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
public class DefaultConnectionListener implements ConnectionListener {

2
src/main/java/org/redisson/connection/LoadBalancer.java
Unescape Escape View File

@ -19,6 +19,6 @@ import java.util.List;
public interface LoadBalancer {
SubscribesConnectionEntry getEntry(List<SubscribesConnectionEntry> clientsCopy);
ClientConnectionsEntry getEntry(List<ClientConnectionsEntry> clientsCopy);
}

4
src/main/java/org/redisson/connection/LoadBalancerManager.java
Unescape Escape View File

@ -20,7 +20,7 @@ import java.util.Collection;
import org.redisson.client.RedisConnection;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import io.netty.util.concurrent.Future;
@ -38,7 +38,7 @@ public interface LoadBalancerManager {
Collection<RedisPubSubConnection> freeze(String host, int port, FreezeReason freezeReason);
void add(SubscribesConnectionEntry entry);
void add(ClientConnectionsEntry entry);
Future<RedisConnection> nextConnection();

22
src/main/java/org/redisson/connection/LoadBalancerManagerImpl.java
Unescape Escape View File

@ -26,7 +26,7 @@ import org.redisson.MasterSlaveServersConfig;
import org.redisson.client.RedisConnection;
import org.redisson.client.RedisConnectionException;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.ConnectionPool;
import org.redisson.misc.PubSubConnectionPoll;
import org.slf4j.Logger;
@ -40,7 +40,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
private final Logger log = LoggerFactory.getLogger(getClass());
private final ConnectionManager connectionManager;
private final Map<InetSocketAddress, SubscribesConnectionEntry> addr2Entry = PlatformDependent.newConcurrentHashMap();
private final Map<InetSocketAddress, ClientConnectionsEntry> addr2Entry = PlatformDependent.newConcurrentHashMap();
private final PubSubConnectionPoll pubSubEntries;
private final ConnectionPool<RedisConnection> entries;
@ -50,7 +50,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
pubSubEntries = new PubSubConnectionPoll(config, connectionManager, entry);
}
public synchronized void add(SubscribesConnectionEntry entry) {
public synchronized void add(ClientConnectionsEntry entry) {
addr2Entry.put(entry.getClient().getAddr(), entry);
entries.add(entry);
pubSubEntries.add(entry);
@ -58,7 +58,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
public int getAvailableClients() {
int count = 0;
for (SubscribesConnectionEntry connectionEntry : addr2Entry.values()) {
for (ClientConnectionsEntry connectionEntry : addr2Entry.values()) {
if (!connectionEntry.isFreezed()) {
count++;
}
@ -68,7 +68,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
public boolean unfreeze(String host, int port, FreezeReason freezeReason) {
InetSocketAddress addr = new InetSocketAddress(host, port);
SubscribesConnectionEntry entry = addr2Entry.get(addr);
ClientConnectionsEntry entry = addr2Entry.get(addr);
if (entry == null) {
throw new IllegalStateException("Can't find " + addr + " in slaves!");
}
@ -90,7 +90,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
public Collection<RedisPubSubConnection> freeze(String host, int port, FreezeReason freezeReason) {
InetSocketAddress addr = new InetSocketAddress(host, port);
SubscribesConnectionEntry connectionEntry = addr2Entry.get(addr);
ClientConnectionsEntry connectionEntry = addr2Entry.get(addr);
if (connectionEntry == null) {
return Collections.emptyList();
}
@ -133,7 +133,7 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
}
public Future<RedisConnection> getConnection(InetSocketAddress addr) {
SubscribesConnectionEntry entry = addr2Entry.get(addr);
ClientConnectionsEntry entry = addr2Entry.get(addr);
if (entry != null) {
return entries.get(entry);
}
@ -146,23 +146,23 @@ public class LoadBalancerManagerImpl implements LoadBalancerManager {
}
public void returnSubscribeConnection(RedisPubSubConnection connection) {
SubscribesConnectionEntry entry = addr2Entry.get(connection.getRedisClient().getAddr());
ClientConnectionsEntry entry = addr2Entry.get(connection.getRedisClient().getAddr());
pubSubEntries.returnConnection(entry, connection);
}
public void returnConnection(RedisConnection connection) {
SubscribesConnectionEntry entry = addr2Entry.get(connection.getRedisClient().getAddr());
ClientConnectionsEntry entry = addr2Entry.get(connection.getRedisClient().getAddr());
entries.returnConnection(entry, connection);
}
public void shutdown() {
for (SubscribesConnectionEntry entry : addr2Entry.values()) {
for (ClientConnectionsEntry entry : addr2Entry.values()) {
entry.getClient().shutdown();
}
}
public void shutdownAsync() {
for (SubscribesConnectionEntry entry : addr2Entry.values()) {
for (ClientConnectionsEntry entry : addr2Entry.values()) {
connectionManager.shutdownAsync(entry.getClient());
}
}

2
src/main/java/org/redisson/connection/MasterSlaveConnectionManager.java
Unescape Escape View File

@ -39,7 +39,7 @@ import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.RedisCommand;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.cluster.ClusterSlotRange;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.InfinitySemaphoreLatch;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

12
src/main/java/org/redisson/connection/MasterSlaveEntry.java
Unescape Escape View File

@ -26,8 +26,8 @@ import org.redisson.client.RedisClient;
import org.redisson.client.RedisConnection;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.cluster.ClusterSlotRange;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
import org.redisson.misc.MasterConnectionPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -45,7 +45,7 @@ public class MasterSlaveEntry {
final Logger log = LoggerFactory.getLogger(getClass());
LoadBalancerManager slaveBalancer;
SubscribesConnectionEntry masterEntry;
ClientConnectionsEntry masterEntry;
final ConnectionListener connectListener;
@ -80,7 +80,7 @@ public class MasterSlaveEntry {
public void setupMasterEntry(String host, int port) {
RedisClient client = connectionManager.createClient(host, port);
masterEntry = new SubscribesConnectionEntry(client, config.getMasterConnectionPoolSize(), 0, connectListener, NodeType.MASTER);
masterEntry = new ClientConnectionsEntry(client, config.getMasterConnectionPoolSize(), 0, connectListener, NodeType.MASTER);
writeConnectionHolder.add(masterEntry);
}
@ -101,7 +101,7 @@ public class MasterSlaveEntry {
private void addSlave(String host, int port, boolean freezed, NodeType mode) {
RedisClient client = connectionManager.createClient(host, port);
SubscribesConnectionEntry entry = new SubscribesConnectionEntry(client,
ClientConnectionsEntry entry = new ClientConnectionsEntry(client,
this.config.getSlaveConnectionPoolSize(),
this.config.getSlaveSubscriptionConnectionPoolSize(), connectListener, mode);
if (freezed) {
@ -133,7 +133,7 @@ public class MasterSlaveEntry {
*
*/
public void changeMaster(String host, int port) {
SubscribesConnectionEntry oldMaster = masterEntry;
ClientConnectionsEntry oldMaster = masterEntry;
setupMasterEntry(host, port);
writeConnectionHolder.remove(oldMaster);
if (slaveBalancer.getAvailableClients() > 1) {

2
src/main/java/org/redisson/connection/RandomLoadBalancer.java
Unescape Escape View File

@ -23,7 +23,7 @@ public class RandomLoadBalancer implements LoadBalancer {
private final Random random = new SecureRandom();
public SubscribesConnectionEntry getEntry(List<SubscribesConnectionEntry> clientsCopy) {
public ClientConnectionsEntry getEntry(List<ClientConnectionsEntry> clientsCopy) {
int ind = random.nextInt(clientsCopy.size());
return clientsCopy.get(ind);
}

2
src/main/java/org/redisson/connection/RoundRobinLoadBalancer.java
Unescape Escape View File

@ -23,7 +23,7 @@ public class RoundRobinLoadBalancer implements LoadBalancer {
private final AtomicInteger index = new AtomicInteger(-1);
@Override
public SubscribesConnectionEntry getEntry(List<SubscribesConnectionEntry> clientsCopy) {
public ClientConnectionsEntry getEntry(List<ClientConnectionsEntry> clientsCopy) {
int ind = Math.abs(index.incrementAndGet() % clientsCopy.size());
return clientsCopy.get(ind);
}

2
src/main/java/org/redisson/connection/SentinelConnectionManager.java
Unescape Escape View File

@ -33,7 +33,7 @@ import org.redisson.client.RedisPubSubConnection;
import org.redisson.client.codec.StringCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.misc.URIBuilder;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

6
src/main/java/org/redisson/connection/SingleEntry.java
Unescape Escape View File

@ -23,7 +23,7 @@ import org.redisson.client.RedisClient;
import org.redisson.client.RedisConnection;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.cluster.ClusterSlotRange;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
import org.redisson.misc.ConnectionPool;
import org.redisson.misc.PubSubConnectionPoll;
@ -36,7 +36,7 @@ public class SingleEntry extends MasterSlaveEntry {
public SingleEntry(Set<ClusterSlotRange> slotRanges, ConnectionManager connectionManager, MasterSlaveServersConfig config, ConnectionListener connectListener) {
super(slotRanges, connectionManager, config, connectListener);
pubSubConnectionHolder = new PubSubConnectionPoll(config, connectionManager, this) {
protected SubscribesConnectionEntry getEntry() {
protected ClientConnectionsEntry getEntry() {
return entries.get(0);
}
};
@ -45,7 +45,7 @@ public class SingleEntry extends MasterSlaveEntry {
@Override
public void setupMasterEntry(String host, int port) {
RedisClient masterClient = connectionManager.createClient(host, port);
masterEntry = new SubscribesConnectionEntry(masterClient,
masterEntry = new ClientConnectionsEntry(masterClient,
config.getMasterConnectionPoolSize(), config.getSlaveSubscriptionConnectionPoolSize(), connectListener, NodeType.MASTER);
writeConnectionHolder.add(masterEntry);
pubSubConnectionHolder.add(masterEntry);

48
src/main/java/org/redisson/misc/ConnectionPool.java
Unescape Escape View File

@ -28,9 +28,9 @@ import org.redisson.client.RedisConnectionException;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.connection.ConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.SubscribesConnectionEntry;
import org.redisson.connection.SubscribesConnectionEntry.FreezeReason;
import org.redisson.connection.SubscribesConnectionEntry.NodeType;
import org.redisson.connection.ClientConnectionsEntry;
import org.redisson.connection.ClientConnectionsEntry.FreezeReason;
import org.redisson.connection.ClientConnectionsEntry.NodeType;
import io.netty.util.Timeout;
import io.netty.util.TimerTask;
@ -40,7 +40,7 @@ import io.netty.util.concurrent.Promise;
public class ConnectionPool<T extends RedisConnection> {
protected final List<SubscribesConnectionEntry> entries = new CopyOnWriteArrayList<SubscribesConnectionEntry>();
protected final List<ClientConnectionsEntry> entries = new CopyOnWriteArrayList<ClientConnectionsEntry>();
final Deque<Promise<T>> promises = new LinkedBlockingDeque<Promise<T>>();
@ -67,7 +67,7 @@ public class ConnectionPool<T extends RedisConnection> {
// }, 1, 1, TimeUnit.SECONDS);
}
public void add(final SubscribesConnectionEntry entry) {
public void add(final ClientConnectionsEntry entry) {
initConnections(entry, new Runnable() {
@Override
public void run() {
@ -77,7 +77,7 @@ public class ConnectionPool<T extends RedisConnection> {
}, true);
}
private void initConnections(final SubscribesConnectionEntry entry, final Runnable runnable, boolean checkFreezed) {
private void initConnections(final ClientConnectionsEntry entry, final Runnable runnable, boolean checkFreezed) {
int minimumIdleSize = getMinimumIdleSize(entry);
if (minimumIdleSize == 0) {
@ -110,21 +110,21 @@ public class ConnectionPool<T extends RedisConnection> {
}
}
protected int getMinimumIdleSize(SubscribesConnectionEntry entry) {
protected int getMinimumIdleSize(ClientConnectionsEntry entry) {
return config.getSlaveConnectionMinimumIdleSize();
}
public void remove(SubscribesConnectionEntry entry) {
public void remove(ClientConnectionsEntry entry) {
entries.remove(entry);
}
protected SubscribesConnectionEntry getEntry() {
protected ClientConnectionsEntry getEntry() {
return config.getLoadBalancer().getEntry(entries);
}
public Future<T> get() {
for (int j = entries.size() - 1; j >= 0; j--) {
SubscribesConnectionEntry entry = getEntry();
ClientConnectionsEntry entry = getEntry();
if (!entry.isFreezed() && tryAcquireConnection(entry)) {
Promise<T> promise = connectionManager.newPromise();
connect(entry, promise);
@ -137,7 +137,7 @@ public class ConnectionPool<T extends RedisConnection> {
return promise;
}
public Future<T> get(SubscribesConnectionEntry entry) {
public Future<T> get(ClientConnectionsEntry entry) {
if (((entry.getNodeType() == NodeType.MASTER && entry.getFreezeReason() == FreezeReason.SYSTEM) || !entry.isFreezed())
&& tryAcquireConnection(entry)) {
Promise<T> promise = connectionManager.newPromise();
@ -150,19 +150,19 @@ public class ConnectionPool<T extends RedisConnection> {
return connectionManager.newFailedFuture(exception);
}
protected boolean tryAcquireConnection(SubscribesConnectionEntry entry) {
protected boolean tryAcquireConnection(ClientConnectionsEntry entry) {
return entry.getFailedAttempts() < config.getSlaveFailedAttempts() && entry.tryAcquireConnection();
}
protected T poll(SubscribesConnectionEntry entry) {
protected T poll(ClientConnectionsEntry entry) {
return (T) entry.pollConnection();
}
protected Future<T> connect(SubscribesConnectionEntry entry) {
protected Future<T> connect(ClientConnectionsEntry entry) {
return (Future<T>) entry.connect(config);
}
private void connect(final SubscribesConnectionEntry entry, final Promise<T> promise) {
private void connect(final ClientConnectionsEntry entry, final Promise<T> promise) {
T conn = poll(entry);
if (conn != null) {
if (!conn.isActive()) {
@ -196,7 +196,7 @@ public class ConnectionPool<T extends RedisConnection> {
});
}
private void promiseSuccessful(final SubscribesConnectionEntry entry, final Promise<T> promise, T conn) {
private void promiseSuccessful(final ClientConnectionsEntry entry, final Promise<T> promise, T conn) {
entry.resetFailedAttempts();
if (!promise.trySuccess(conn)) {
releaseConnection(entry, conn);
@ -204,7 +204,7 @@ public class ConnectionPool<T extends RedisConnection> {
}
}
private void promiseFailure(SubscribesConnectionEntry entry, Promise<T> promise, Throwable cause) {
private void promiseFailure(ClientConnectionsEntry entry, Promise<T> promise, Throwable cause) {
if (entry.incFailedAttempts() == config.getSlaveFailedAttempts()) {
if (entry.getNodeType() == NodeType.SLAVE) {
connectionManager.slaveDown(masterSlaveEntry, entry.getClient().getAddr().getHostName(),
@ -219,14 +219,14 @@ public class ConnectionPool<T extends RedisConnection> {
promise.tryFailure(cause);
}
private void freezeMaster(SubscribesConnectionEntry entry) {
private void freezeMaster(ClientConnectionsEntry entry) {
if (entry.freezeMaster(FreezeReason.RECONNECT)) {
scheduleCheck(entry);
}
}
private void promiseFailure(SubscribesConnectionEntry entry, Promise<T> promise, T conn) {
private void promiseFailure(ClientConnectionsEntry entry, Promise<T> promise, T conn) {
int attempts = entry.incFailedAttempts();
if (attempts == config.getSlaveFailedAttempts()) {
if (entry.getNodeType() == NodeType.SLAVE) {
@ -247,7 +247,7 @@ public class ConnectionPool<T extends RedisConnection> {
promise.tryFailure(cause);
}
private void scheduleCheck(final SubscribesConnectionEntry entry) {
private void scheduleCheck(final ClientConnectionsEntry entry) {
connectionManager.newTimeout(new TimerTask() {
@Override
public void run(Timeout timeout) throws Exception {
@ -321,7 +321,7 @@ public class ConnectionPool<T extends RedisConnection> {
}, config.getSlaveReconnectionTimeout(), TimeUnit.MILLISECONDS);
}
public void returnConnection(SubscribesConnectionEntry entry, T connection) {
public void returnConnection(ClientConnectionsEntry entry, T connection) {
if (entry.isFreezed()) {
connection.closeAsync();
} else {
@ -333,13 +333,13 @@ public class ConnectionPool<T extends RedisConnection> {
releaseConnection(entry);
}
protected void releaseConnection(SubscribesConnectionEntry entry) {
protected void releaseConnection(ClientConnectionsEntry entry) {
entry.releaseConnection();
handleQueue(entry, true);
}
private void handleQueue(SubscribesConnectionEntry entry, boolean checkFreezed) {
private void handleQueue(ClientConnectionsEntry entry, boolean checkFreezed) {
while (true) {
if (checkFreezed && entry.isFreezed()) {
return;
@ -361,7 +361,7 @@ public class ConnectionPool<T extends RedisConnection> {
}
}
protected void releaseConnection(SubscribesConnectionEntry entry, T conn) {
protected void releaseConnection(ClientConnectionsEntry entry, T conn) {
entry.releaseConnection(conn);
}

6
src/main/java/org/redisson/misc/MasterConnectionPool.java
Unescape Escape View File

@ -19,7 +19,7 @@ import org.redisson.MasterSlaveServersConfig;
import org.redisson.client.RedisConnection;
import org.redisson.connection.ConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.SubscribesConnectionEntry;
import org.redisson.connection.ClientConnectionsEntry;
public class MasterConnectionPool extends ConnectionPool<RedisConnection> {
@ -29,12 +29,12 @@ public class MasterConnectionPool extends ConnectionPool<RedisConnection> {
}
@Override
protected SubscribesConnectionEntry getEntry() {
protected ClientConnectionsEntry getEntry() {
return entries.get(0);
}
@Override
protected int getMinimumIdleSize(SubscribesConnectionEntry entry) {
protected int getMinimumIdleSize(ClientConnectionsEntry entry) {
return config.getMasterConnectionMinimumIdleSize();
}

14
src/main/java/org/redisson/misc/PubSubConnectionPoll.java
Unescape Escape View File

@ -19,7 +19,7 @@ import org.redisson.MasterSlaveServersConfig;
import org.redisson.client.RedisPubSubConnection;
import org.redisson.connection.ConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.connection.SubscribesConnectionEntry;
import org.redisson.connection.ClientConnectionsEntry;
import io.netty.util.concurrent.Future;
@ -30,32 +30,32 @@ public class PubSubConnectionPoll extends ConnectionPool<RedisPubSubConnection>
}
@Override
protected RedisPubSubConnection poll(SubscribesConnectionEntry entry) {
protected RedisPubSubConnection poll(ClientConnectionsEntry entry) {
return entry.pollSubscribeConnection();
}
@Override
protected int getMinimumIdleSize(SubscribesConnectionEntry entry) {
protected int getMinimumIdleSize(ClientConnectionsEntry entry) {
return config.getSlaveSubscriptionConnectionMinimumIdleSize();
}
@Override
protected Future<RedisPubSubConnection> connect(SubscribesConnectionEntry entry) {
protected Future<RedisPubSubConnection> connect(ClientConnectionsEntry entry) {
return entry.connectPubSub(config);
}
@Override
protected boolean tryAcquireConnection(SubscribesConnectionEntry entry) {
protected boolean tryAcquireConnection(ClientConnectionsEntry entry) {
return entry.tryAcquireSubscribeConnection();
}
@Override
protected void releaseConnection(SubscribesConnectionEntry entry) {
protected void releaseConnection(ClientConnectionsEntry entry) {
entry.releaseSubscribeConnection();
}
@Override
protected void releaseConnection(SubscribesConnectionEntry entry, RedisPubSubConnection conn) {
protected void releaseConnection(ClientConnectionsEntry entry, RedisPubSubConnection conn) {
entry.releaseSubscribeConnection(conn);
}

Write Preview
Loading…
Cancel
Save