public Flowable<V> takeElements() {
        return ElementsStream.takeElements(queue::takeAsync);
    }
@Test
    public void testTakeElements() throws InterruptedException {
        RBlockingQueueRx<Integer> queue = redisson.getBlockingQueue("test");
        List<Integer> elements = new ArrayList<>();
        queue.takeElements().subscribe(new Subscriber<Integer>() {

            @Override
            public void onSubscribe(Subscription s) {
                s.request(4);
            }

            @Override
            public void onNext(Integer t) {
                elements.add(t);
            }

            @Override
            public void onError(Throwable t) {
            }

            @Override
            public void onComplete() {
            }
        });
        
        for (int i = 0; i < 10; i++) {
            sync(queue.add(i));
        }

        Thread.sleep(500);

        assertThat(elements).containsExactly(0, 1, 2, 3);
    }