public Optional<ByteBuffer> dequeue() throws QueueException {
        if (!currentHeadPtr.isGreaterThan(currentTailPtr)) {
            if (currentTailPtr.isGreaterThan(currentHeadPtr)) {
                // sanity check
                throw new QueueException("Current tail " + currentTailPtr + " is forward head " + currentHeadPtr);
            }
            // head and tail pointer are the same, the queue is empty
            return Optional.empty();
        }
        if (tailSegment == null) {
            tailSegment = queuePool.openNextTailSegment(name).get();
        }

        LOG.debug("currentTail is {}", currentTailPtr);
        if (containsHeader(tailSegment, currentTailPtr)) {
            // currentSegment contains at least the header (payload length)
            final VirtualPointer existingTail;
            if (isTailFirstUsage(currentTailPtr)) {
                // move to the first readable byte
                existingTail = currentTailPtr.plus(1);
            } else {
                existingTail = currentTailPtr.copy();
            }
            final int payloadLength = tailSegment.readHeader(existingTail);
            // tail must be moved to the next byte to read, so has to move to
            // header size + payload size + 1
            final int fullMessageSize = payloadLength + LENGTH_HEADER_SIZE;
            long remainingInSegment = tailSegment.bytesAfter(existingTail) + 1;
            if (remainingInSegment > fullMessageSize) {
                // tail segment fully contains the payload with space left over
                currentTailPtr = existingTail.moveForward(fullMessageSize);
                // read data from currentTail + 4 bytes(the length)
                final VirtualPointer dataStart = existingTail.moveForward(LENGTH_HEADER_SIZE);

                return Optional.of(readData(tailSegment, dataStart, payloadLength));
            } else {
                // payload is split across currentSegment and next ones
                VirtualPointer dataStart = existingTail.moveForward(LENGTH_HEADER_SIZE);

                if (remainingInSegment - LENGTH_HEADER_SIZE == 0) {
                    queuePool.consumedTailSegment(name);
                    if (QueuePool.queueDebug) {
                        tailSegment.fillWith((byte) 'D');
                    }
                    tailSegment = queuePool.openNextTailSegment(name).get();
                }

                LOG.debug("Loading payload size {}", payloadLength);
                return Optional.of(loadPayloadFromSegments(payloadLength, tailSegment, dataStart));
            }
        } else {
            // header is split across 2 segments
            // the currentSegment is still the tailSegment
            // read the length header that's crossing 2 segments
            final CrossSegmentHeaderResult result = decodeCrossHeader(tailSegment, currentTailPtr);

            // load all payload parts from the segments
            LOG.debug("Loading payload size {}", result.payloadLength);
            return Optional.of(loadPayloadFromSegments(result.payloadLength, result.segment, result.pointer));
        }
    }
@Test
    public void readFromEmptyQueue() throws QueueException {
        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder, PAGE_SIZE, SEGMENT_SIZE);
        final Queue queue = queuePool.getOrCreate("test");

        assertFalse(queue.dequeue().isPresent(), "Pulling from empty queue MUST return null value");
    }