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

refactoring

Browse Source
pull/4061/head
Nikita Koksharov 3 years ago
parent 132e34c757
commit cbb70e2f70
19 changed files with 159 additions and 187 deletions
Show Stats Download Patch File Download Diff File

41
redisson-spring-data/redisson-spring-data-21/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,15 +15,12 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -147,15 +144,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -163,9 +156,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -178,15 +176,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -194,9 +188,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

41
redisson-spring-data/redisson-spring-data-22/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,15 +15,12 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -146,15 +143,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -162,9 +155,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -177,15 +175,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -193,9 +187,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

41
redisson-spring-data/redisson-spring-data-23/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,15 +15,12 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -146,15 +143,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -162,9 +155,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -177,15 +175,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -193,9 +187,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

41
redisson-spring-data/redisson-spring-data-24/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,15 +15,12 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -146,15 +143,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -162,9 +155,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -177,15 +175,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -193,9 +187,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

42
redisson-spring-data/redisson-spring-data-25/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,15 +15,12 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -36,7 +33,6 @@ import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@ -147,15 +143,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -163,9 +155,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -178,15 +175,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -194,9 +187,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

42
redisson-spring-data/redisson-spring-data-26/src/main/java/org/redisson/spring/data/connection/RedissonReactiveSubscription.java
Unescape Escape View File

@ -15,7 +15,6 @@
*/
package org.redisson.spring.data.connection;
import org.redisson.api.RFuture;
import org.redisson.client.BaseRedisPubSubListener;
import org.redisson.client.ChannelName;
import org.redisson.client.RedisPubSubListener;
@ -23,8 +22,6 @@ import org.redisson.client.codec.ByteArrayCodec;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubType;
import org.redisson.connection.ConnectionManager;
import org.redisson.misc.CountableListener;
import org.redisson.misc.RedissonPromise;
import org.redisson.pubsub.PubSubConnectionEntry;
import org.redisson.pubsub.PublishSubscribeService;
import org.springframework.data.redis.connection.ReactiveSubscription;
@ -39,7 +36,6 @@ import java.util.*;
import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
@ -175,15 +171,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> unsubscribe(ByteBuffer... channels) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, channels.length);
List<CompletableFuture<?>> futures = new ArrayList<>(channels.length);
for (ByteBuffer channel : channels) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) {
PubSubConnectionEntry entry = RedissonReactiveSubscription.this.channels.get(cn);
if (!entry.hasListeners(cn)) {
@ -191,9 +183,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}
@ -206,15 +203,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public Mono<Void> pUnsubscribe(ByteBuffer... patterns) {
monosListener.acquire();
return Mono.defer(() -> {
RedissonPromise<Void> result = new RedissonPromise<>();
result.onComplete((r, ex) -> {
monosListener.release();
});
CountableListener<Void> listener = new CountableListener<>(result, null, patterns.length);
List<CompletableFuture<?>> futures = new ArrayList<>(patterns.length);
for (ByteBuffer channel : patterns) {
ChannelName cn = toChannelName(channel);
RFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f.onComplete((res, e) -> {
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn);
entries.stream()
@ -222,9 +215,14 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
}
});
f.onComplete(listener);
futures.add(f);
}
return Mono.fromFuture(result);
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
future = future.whenComplete((r, e) -> {
monosListener.release();
});
return Mono.fromFuture(future);
});
}

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

@ -75,7 +75,7 @@ public class ElementsSubscribeService {
subscribeListeners.put(listenerId, f);
}
f.onComplete((r, e) -> {
f.whenComplete((r, e) -> {
if (e != null) {
connectionManager.newTimeout(t -> {
resubscribe(func, consumer);

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

@ -62,7 +62,7 @@ public class MapWriteBehindTask {
private void pollTask(Map<Object, Object> addedMap, List<Object> deletedKeys) {
RFuture<MapWriterTask> future = writeBehindTasks.pollAsync();
future.onComplete((task, e) -> {
future.whenComplete((task, e) -> {
if (e != null) {
log.error(e.getMessage(), e);

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

@ -140,7 +140,7 @@ public abstract class QueueTransferTask {
private void pushTask() {
RFuture<Long> startTimeFuture = pushTaskAsync();
startTimeFuture.onComplete((res, e) -> {
startTimeFuture.whenComplete((res, e) -> {
if (e != null) {
if (e instanceof RedissonShutdownException) {
return;

2
redisson/src/main/java/org/redisson/client/RedisConnection.java
Unescape Escape View File

@ -289,7 +289,7 @@ public class RedisConnection implements RedisCommands {
channel.close();
} else {
RFuture<Void> f = async(RedisCommands.QUIT);
f.onComplete((res, e) -> {
f.whenComplete((res, e) -> {
channel.close();
});
}

2
redisson/src/main/java/org/redisson/client/handler/BaseConnectionHandler.java
Unescape Escape View File

@ -97,7 +97,7 @@ public abstract class BaseConnectionHandler<C extends RedisConnection> extends C
final AtomicBoolean retry = new AtomicBoolean();
final AtomicInteger commandsCounter = new AtomicInteger(futures.size());
for (RFuture<Object> future : futures) {
future.onComplete((res, e) -> {
future.whenComplete((res, e) -> {
if (e != null) {
if (e instanceof RedisLoadingException) {
if (retry.compareAndSet(false, true)) {

45
redisson/src/main/java/org/redisson/command/CommandAsyncService.java
Unescape Escape View File

@ -521,7 +521,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
if (e != null) {
if (e.getMessage().startsWith("NOSCRIPT")) {
RFuture<String> loadFuture = loadScript(executor.getRedisClient(), script);
loadFuture.onComplete((r, ex) -> {
loadFuture.whenComplete((r, ex) -> {
if (ex != null) {
free(pps);
mainPromise.completeExceptionally(ex);
@ -627,29 +627,7 @@ public class CommandAsyncService implements CommandAsyncExecutor {
return connectionManager.calcSlot(k);
}, Collectors.toList())));
long total = entry2keys.values().stream().mapToInt(m -> m.size()).sum();
RPromise<R> result = new RedissonPromise<>();
AtomicLong executed = new AtomicLong(total);
AtomicReference<Throwable> failed = new AtomicReference<>();
BiConsumer<T, Throwable> listener = (res, ex) -> {
if (ex != null) {
failed.set(ex);
} else {
if (res != null) {
callback.onSlotResult(res);
}
}
if (executed.decrementAndGet() == 0) {
if (failed.get() != null) {
result.tryFailure(failed.get());
} else {
result.trySuccess(callback.onFinish());
}
}
};
List<CompletableFuture<?>> futures = new ArrayList<>();
for (Entry<MasterSlaveEntry, Map<Integer, List<String>>> entry : entry2keys.entrySet()) {
// executes in batch due to CROSSLOT error
CommandBatchService executorService;
@ -667,10 +645,10 @@ public class CommandAsyncService implements CommandAsyncExecutor {
}
if (readOnly) {
RFuture<T> f = executorService.readAsync(entry.getKey(), codec, c, callback.createParams(groupedKeys));
f.onComplete(listener);
futures.add(f.toCompletableFuture());
} else {
RFuture<T> f = executorService.writeAsync(entry.getKey(), codec, c, callback.createParams(groupedKeys));
f.onComplete(listener);
futures.add(f.toCompletableFuture());
}
}
@ -679,7 +657,16 @@ public class CommandAsyncService implements CommandAsyncExecutor {
}
}
return result;
CompletableFuture<Void> future = CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]));
CompletableFuture<R> result = future.whenComplete((res, e) -> {
futures.forEach(f -> {
if (!f.isCompletedExceptionally() && f.getNow(null) != null) {
callback.onSlotResult((T) f.getNow(null));
}
});
}).thenApply(r -> callback.onFinish());
return new CompletableFutureWrapper<>(result);
}
@ -760,9 +747,9 @@ public class CommandAsyncService implements CommandAsyncExecutor {
private <V> void poll(String name, Codec codec, RPromise<V> result, AtomicReference<Iterator<String>> ref,
List<String> names, AtomicLong counter, RedisCommand<Object> command) {
if (ref.get().hasNext()) {
String currentName = ref.get().next().toString();
String currentName = ref.get().next();
RFuture<V> future = writeAsync(currentName, codec, command, currentName, 1);
future.onComplete((res, e) -> {
future.whenComplete((res, e) -> {
if (e != null) {
result.tryFailure(e);
return;

15
redisson/src/main/java/org/redisson/command/CommandBatchService.java
Unescape Escape View File

@ -202,16 +202,9 @@ public class CommandBatchService extends CommandAsyncService {
}
public RFuture<Void> executeAsyncVoid() {
RedissonPromise<Void> promise = new RedissonPromise<Void>();
RFuture<BatchResult<?>> resFuture = executeAsync();
resFuture.onComplete((res, e) -> {
if (e == null) {
promise.trySuccess(null);
} else {
promise.tryFailure(e);
}
});
return promise;
CompletableFuture<BatchResult<?>> resFuture = executeAsync().toCompletableFuture();
CompletableFuture<Void> s = resFuture.thenApply(res -> null);
return new CompletableFutureWrapper<>(s);
}
public boolean isExecuted() {
@ -343,7 +336,7 @@ public class CommandBatchService extends CommandAsyncService {
service.executeAsync();
}
entry.getKey().onComplete((res, e) -> {
entry.getKey().whenComplete((res, e) -> {
handle(voidPromise, slots, entry.getKey());
});
}

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

@ -73,7 +73,7 @@ public class RedisClientEntry implements ClusterNode {
public RFuture<Boolean> pingAsync(long timeout, TimeUnit timeUnit) {
RPromise<Boolean> result = new RedissonPromise<>();
RFuture<Boolean> f = commandExecutor.readAsync(client, null, RedisCommands.PING_BOOL);
f.onComplete((res, e) -> {
f.whenComplete((res, e) -> {
if (e != null) {
result.trySuccess(false);
return;

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

@ -175,7 +175,7 @@ public class ReplicatedConnectionManager extends MasterSlaveConnectionManager {
}
RFuture<Map<String, String>> result = connection.async(RedisCommands.INFO_REPLICATION);
result.onComplete((r, ex) -> {
result.whenComplete((r, ex) -> {
if (ex != null) {
log.error(ex.getMessage(), ex);
closeNodeConnection(connection);

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

@ -369,7 +369,7 @@ abstract class ConnectionPool<T extends RedisConnection> {
}
RFuture<String> f = c.async(RedisCommands.PING);
f.onComplete((t, ex) -> {
f.whenComplete((t, ex) -> {
try {
synchronized (entry) {
if (entry.getFreezeReason() != FreezeReason.RECONNECT) {

2
redisson/src/main/java/org/redisson/eviction/EvictionTask.java
Unescape Escape View File

@ -72,7 +72,7 @@ abstract class EvictionTask implements Runnable {
}
RFuture<Integer> future = execute();
future.onComplete((size, e) -> {
future.whenComplete((size, e) -> {
if (e != null) {
schedule();
return;

18
redisson/src/main/java/org/redisson/pubsub/PublishSubscribeService.java
Unescape Escape View File

@ -18,7 +18,6 @@ package org.redisson.pubsub;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelFutureListener;
import org.redisson.PubSubPatternStatusListener;
import org.redisson.api.RFuture;
import org.redisson.client.*;
import org.redisson.client.codec.Codec;
import org.redisson.client.protocol.pubsub.PubSubStatusMessage;
@ -27,7 +26,6 @@ import org.redisson.config.MasterSlaveServersConfig;
import org.redisson.connection.ConnectionManager;
import org.redisson.connection.MasterSlaveEntry;
import org.redisson.misc.RPromise;
import org.redisson.misc.RedissonPromise;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -441,22 +439,22 @@ public class PublishSubscribeService {
entry2PubSubConnection.remove(entry);
}
public RFuture<Codec> unsubscribe(ChannelName channelName, PubSubType topicType) {
public CompletableFuture<Codec> unsubscribe(ChannelName channelName, PubSubType topicType) {
return unsubscribe(channelName, getEntry(channelName), topicType);
}
private RFuture<Codec> unsubscribe(ChannelName channelName, MasterSlaveEntry e, PubSubType topicType) {
private CompletableFuture<Codec> unsubscribe(ChannelName channelName, MasterSlaveEntry e, PubSubType topicType) {
if (connectionManager.isShuttingDown()) {
return RedissonPromise.newSucceededFuture(null);
return CompletableFuture.completedFuture(null);
}
RPromise<Codec> result = new RedissonPromise<>();
CompletableFuture<Codec> result = new CompletableFuture<>();
AsyncSemaphore lock = getSemaphore(channelName);
lock.acquire(() -> {
PubSubConnectionEntry entry = name2PubSubConnection.remove(new PubSubKey(channelName, e));
if (entry == null) {
lock.release();
result.trySuccess(null);
result.complete(null);
return;
}
@ -482,7 +480,7 @@ public class PublishSubscribeService {
executed.set(true);
lock.release();
result.trySuccess(entryCodec);
result.complete(entryCodec);
return true;
}
return false;
@ -572,12 +570,12 @@ public class PublishSubscribeService {
}
private void reattachPubSubListeners(ChannelName channelName, MasterSlaveEntry en, Collection<RedisPubSubListener<?>> listeners, PubSubType topicType) {
RFuture<Codec> subscribeCodecFuture = unsubscribe(channelName, en, topicType);
CompletableFuture<Codec> subscribeCodecFuture = unsubscribe(channelName, en, topicType);
if (listeners.isEmpty()) {
return;
}
subscribeCodecFuture.onComplete((subscribeCodec, e) -> {
subscribeCodecFuture.whenComplete((subscribeCodec, e) -> {
if (subscribeCodec == null) {
return;
}

2
redisson/src/test/java/org/redisson/RedissonBatchTest.java
Unescape Escape View File

@ -365,6 +365,8 @@ public class RedissonBatchTest extends BaseTest {
String[] t = redisson.getKeys().getKeysStreamByPattern("*").toArray(String[]::new);
} catch (Exception ex) {
ex.printStackTrace();
} finally {
redisson.shutdown();
}
}

Write Preview
Loading…
Cancel
Save