refactoring

redisson/src/main/java/org/redisson/pubsub/PublishSubscribe.java

@@ -44,7 +44,7 @@ abstract class PublishSubscribe<E extends PubSubEntry<E>> {
 
     public void unsubscribe(E entry, String entryName, String channelName) {
         AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
-        semaphore.acquire(() -> {
+        semaphore.acquire().thenAccept(c -> {
             if (entry.release() == 0) {
                 entries.remove(entryName);
                 service.unsubscribe(PubSubType.UNSUBSCRIBE, new ChannelName(channelName))
@@ -55,7 +55,6 @@ abstract class PublishSubscribe<E extends PubSubEntry<E>> {
                 semaphore.release();
             }
         });
-
     }
 
     public void timeout(CompletableFuture<?> promise) {
@@ -70,7 +69,7 @@ abstract class PublishSubscribe<E extends PubSubEntry<E>> {
         AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
         CompletableFuture<E> newPromise = new CompletableFuture<>();
 
-        semaphore.acquire(() -> {
+        semaphore.acquire().thenAccept(c -> {
             if (newPromise.isDone()) {
                 semaphore.release();
                 return;

redisson/src/main/java/org/redisson/pubsub/PublishSubscribeService.java

@@ -197,7 +197,7 @@ public class PublishSubscribeService {
                     "Unable to acquire subscription lock after " + timeout + "ms. " +
                             "Try to increase 'timeout', 'subscriptionsPerConnection', 'subscriptionConnectionPoolSize' parameters."));
         }, timeout, TimeUnit.MILLISECONDS);
-        lock.acquire(() -> {
+        lock.acquire().thenAccept(r -> {
             if (!lockTimeout.cancel() || promise.isDone()) {
                 lock.release();
                 return;
@@ -257,7 +257,7 @@ public class PublishSubscribeService {
             return;
         }
 
-        freePubSubLock.acquire(() -> {
+        freePubSubLock.acquire().thenAccept(c -> {
             if (promise.isDone()) {
                 lock.release();
                 freePubSubLock.release();
@@ -390,7 +390,7 @@ public class PublishSubscribeService {
                 return;
             }
 
-            freePubSubLock.acquire(() -> {
+            freePubSubLock.acquire().thenAccept(c -> {
                 PubSubConnectionEntry entry = new PubSubConnectionEntry(conn, connectionManager);
                 int remainFreeAmount = entry.tryAcquire();
 
@@ -553,7 +553,7 @@ public class PublishSubscribeService {
                     continue;
                 }
 
-                freePubSubLock.acquire(() -> {
+                freePubSubLock.acquire().thenAccept(r -> {
                     e.getValue().getEntries().remove(entry);
                     freePubSubLock.release();
                 });