public CompletableFuture<E> subscribe(String entryName, String channelName) {
        AsyncSemaphore semaphore = service.getSemaphore(new ChannelName(channelName));
        CompletableFuture<E> newPromise = new CompletableFuture<>();

        semaphore.acquire().thenAccept(c -> {
            if (newPromise.isDone()) {
                semaphore.release();
                return;
            }

            E entry = entries.get(entryName);
            if (entry != null) {
                entry.acquire();
                semaphore.release();
                entry.getPromise().whenComplete((r, e) -> {
                    if (e != null) {
                        newPromise.completeExceptionally(e);
                        return;
                    }
                    newPromise.complete(r);
                });
                return;
            }

            E value = createEntry(newPromise);
            value.acquire();

            E oldValue = entries.putIfAbsent(entryName, value);
            if (oldValue != null) {
                oldValue.acquire();
                semaphore.release();
                oldValue.getPromise().whenComplete((r, e) -> {
                    if (e != null) {
                        newPromise.completeExceptionally(e);
                        return;
                    }
                    newPromise.complete(r);
                });
                return;
            }

            RedisPubSubListener<Object> listener = createListener(channelName, value);
            CompletableFuture<PubSubConnectionEntry> s = service.subscribeNoTimeout(LongCodec.INSTANCE, channelName, semaphore, listener);
            newPromise.whenComplete((r, e) -> {
                if (e != null) {
                    s.completeExceptionally(e);
                }
            });
            s.whenComplete((r, e) -> {
                if (e != null) {
                    entries.remove(entryName);
                    value.getPromise().completeExceptionally(e);
                    return;
                }
                if (!value.getPromise().complete(value)) {
                    if (value.getPromise().isCompletedExceptionally()) {
                        entries.remove(entryName);
                    }
                }
            });

        });

        return newPromise;
    }
@Test
    public void testSubscribeForRaceCondition() throws InterruptedException {
        AtomicReference<CompletableFuture<PubSubConnectionEntry>> sRef = new AtomicReference<>();
        new MockUp<PublishSubscribeService>() {

            @Mock
            AsyncSemaphore getSemaphore(ChannelName channelName) {
                return new AsyncSemaphore(1);
            }

            @Mock
            CompletableFuture<PubSubConnectionEntry> subscribeNoTimeout(
                    Codec codec, String channelName,
                    AsyncSemaphore semaphore, RedisPubSubListener<?>... listeners) {
                sRef.set(new CompletableFuture<>());
                return sRef.get();
            }
        };

        CompletableFuture<RedissonLockEntry> newPromise = lockPubSub.subscribe(
                "test", "redisson_lock__channel__test"
        );
        sRef.get().whenComplete((r, e) -> {
            try {
                Thread.sleep(1);
            } catch (InterruptedException ex) {
                throw new RuntimeException(ex);
            }
        });

        Thread thread1 = new Thread(() -> sRef.get().complete(null));
        Thread thread2 = new Thread(() -> newPromise.completeExceptionally(new RedisTimeoutException("test")));

        thread1.start();
        thread2.start();
        thread1.join();
        thread2.join();

        assertTrue(newPromise.isCompletedExceptionally());
        assertTrue(sRef.get().isDone());
        assertFalse(sRef.get().isCompletedExceptionally());

        CompletableFuture<RedissonLockEntry> secondPromise = lockPubSub.subscribe(
                "test", "redisson_lock__channel__test"
        );
        Thread thread3 = new Thread(() -> secondPromise.complete(null));
        thread3.start();
        thread3.join();
        assertTrue(secondPromise.isDone());
        assertFalse(secondPromise.isCompletedExceptionally());
    }