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

Improvement - Virtual Threads compatibility #5499

Browse Source
pull/5520/head
Nikita Koksharov 1 year ago
parent 4d609c061a
commit 89ceaad0be
10 changed files with 248 additions and 323 deletions
Show Stats Download Patch File Download Diff File

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

@ -33,6 +33,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -45,20 +46,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -66,13 +64,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -108,7 +104,6 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
}); });
return Mono.fromFuture(future); return Mono.fromFuture(future);
}); });
} }
protected ChannelName toChannelName(ByteBuffer channel) { protected ChannelName toChannelName(ByteBuffer channel) {
@ -149,17 +144,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -186,12 +177,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -33,6 +33,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -45,20 +46,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -66,13 +64,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -148,17 +144,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -185,12 +177,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -33,6 +33,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -45,20 +46,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -66,13 +64,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -148,17 +144,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -185,12 +177,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -33,6 +33,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -45,20 +46,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -66,13 +64,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -148,17 +144,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -185,12 +177,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -33,6 +33,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -45,20 +46,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -66,13 +64,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -148,17 +144,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -185,12 +177,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -36,6 +36,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -48,20 +49,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -69,13 +67,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -175,17 +171,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -212,12 +204,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -36,6 +36,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -48,20 +49,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -69,13 +67,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -175,17 +171,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -212,12 +204,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -36,6 +36,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -48,20 +49,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -69,13 +67,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -175,17 +171,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -212,12 +204,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -36,6 +36,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -48,20 +49,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -69,13 +67,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -175,17 +171,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -212,12 +204,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

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

@ -36,6 +36,7 @@ import java.util.*;
import java.util.Map.Entry; import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture; import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference; import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors; import java.util.stream.Collectors;
@ -48,20 +49,17 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
public static class ListenableCounter { public static class ListenableCounter {
private int state; private final AtomicInteger state = new AtomicInteger();
private Runnable r; private Runnable r;
public synchronized void acquire() { public void acquire() {
state++; state.incrementAndGet();
} }
public void release() { public void release() {
synchronized (this) { if (state.decrementAndGet() != 0) {
state--;
if (state != 0) {
return; return;
} }
}
if (r != null) { if (r != null) {
r.run(); r.run();
@ -69,13 +67,11 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
public synchronized void addListener(Runnable r) { public void addListener(Runnable r) {
synchronized (this) { if (state.get() != 0) {
if (state != 0) {
this.r = r; this.r = r;
return; return;
} }
}
r.run(); r.run();
} }
@ -107,10 +103,6 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
} }
} }
@Override
public void onPatternMessage(CharSequence pattern, CharSequence channel, Object message) {
}
@Override @Override
public void onMessage(CharSequence channel, Object msg) { public void onMessage(CharSequence channel, Object msg) {
} }
@ -175,17 +167,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.UNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.channels) { RedissonReactiveSubscription.this.channels.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.channels.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
for (PubSubConnectionEntry entry : entries) {
if (!entry.hasListeners(cn)) {
entries.remove(entry);
if (entries.isEmpty()) { if (entries.isEmpty()) {
RedissonReactiveSubscription.this.channels.remove(cn); return null;
}
}
}
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }
@ -212,12 +200,13 @@ public class RedissonReactiveSubscription implements ReactiveSubscription {
ChannelName cn = toChannelName(channel); ChannelName cn = toChannelName(channel);
CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE); CompletableFuture<Codec> f = subscribeService.unsubscribe(cn, PubSubType.PUNSUBSCRIBE);
f = f.whenComplete((res, e) -> { f = f.whenComplete((res, e) -> {
synchronized (RedissonReactiveSubscription.this.patterns) { RedissonReactiveSubscription.this.patterns.computeIfPresent(cn, (key, entries) -> {
Collection<PubSubConnectionEntry> entries = RedissonReactiveSubscription.this.patterns.get(cn); entries.removeIf(entry -> !entry.hasListeners(cn));
entries.stream() if (entries.isEmpty()) {
.filter(en -> en.hasListeners(cn)) return null;
.forEach(ee -> RedissonReactiveSubscription.this.patterns.remove(cn));
} }
return entries;
});
}); });
futures.add(f); futures.add(f);
} }

Write Preview
Loading…
Cancel
Save