@Override
    public int subscribeOnElements(Consumer<V> consumer) {
        return getServiceManager().getElementsSubscribeService()
                .subscribeOnElements(this::takeAsync, consumer);
    }
@Test
    public void testSubscribeOnElements() {
        RBlockingQueue<Integer> q = redisson.getBlockingQueue("test");
        Set<Integer> values = new HashSet<>();
        int listnerId = q.subscribeOnElements(v -> {
            values.add(v);
        });

        for (int i = 0; i < 10; i++) {
            q.add(i);
        }

        Awaitility.await().atMost(Duration.ofSeconds(2)).until(() -> {
            return values.size() == 10;
        });

        q.unsubscribe(listnerId);

        q.add(11);
        q.add(12);

        assertThat(values).hasSize(10);
    }