public void enqueue(ByteBuffer payload) throws QueueException {
        final int messageSize = LENGTH_HEADER_SIZE + payload.remaining();
        if (headSegment.hasSpace(currentHeadPtr, messageSize)) {
            LOG.debug("Head segment has sufficient space for message length {}", LENGTH_HEADER_SIZE + payload.remaining());
            writeData(headSegment, currentHeadPtr.plus(1), payload);
            // move head segment
            currentHeadPtr = currentHeadPtr.moveForward(messageSize);
            return;
        }

        LOG.debug("Head segment doesn't have enough space");
        // the payload can't be fully contained into the current head segment and needs to be splitted
        // with another segment.


        final int dataSize = payload.remaining();
        final ByteBuffer rawData = (ByteBuffer) ByteBuffer.allocate(LENGTH_HEADER_SIZE + dataSize)
            .putInt(dataSize)
            .put(payload)
            .flip();

        // the bytes written from the payload input
        long bytesRemainingInHeaderSegment = Math.min(rawData.remaining(), headSegment.bytesAfter(currentHeadPtr));
        LOG.trace("Writing partial payload to offset {} for {} bytes", currentHeadPtr, bytesRemainingInHeaderSegment);

        if (bytesRemainingInHeaderSegment > 0) {
            int copySize = (int) bytesRemainingInHeaderSegment;
            ByteBuffer slice = rawData.slice();
            slice.limit(copySize);
            writeDataNoHeader(headSegment, currentHeadPtr.plus(1), slice);
            currentHeadPtr = currentHeadPtr.moveForward(bytesRemainingInHeaderSegment);
            // No need to move newSegmentPointer the pointer because the last spinningMove has already moved it

            // shift forward the consumption point
            rawData.position(rawData.position() + copySize);
        }

        Segment newSegment = null;

        // till the payload is not completely stored,
        // save the remaining part into a new segment.
        while (rawData.hasRemaining()) {
            // To request the next segment, it's needed to be done in global lock.
            newSegment = queuePool.nextFreeSegment();
            //notify segment creation for queue in queue pool
            allocationListener.segmentedCreated(name, newSegment);

            int copySize = (int) Math.min(rawData.remaining(), allocator.getSegmentSize());
            ByteBuffer slice = rawData.slice();
            slice.limit(copySize);

            currentHeadPtr = currentHeadPtr.moveForward(copySize);
            writeDataNoHeader(newSegment, newSegment.begin, slice);
            headSegment = newSegment;

            // shift forward the consumption point
            rawData.position(rawData.position() + copySize);
        }
    }
@Test
    public void insertSomeDataIntoNewQueue() throws QueueException, IOException {
        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder, PAGE_SIZE, SEGMENT_SIZE);
        final Queue queue = queuePool.getOrCreate("test");
        queue.enqueue(ByteBuffer.wrap("AAAA".getBytes(StandardCharsets.UTF_8)));

        // verify
        final HashSet<String> fileset = new HashSet<>(Arrays.asList(tempQueueFolder.toFile().list()));
        assertEquals(2, fileset.size());
        assertTrue(fileset.contains("checkpoint.properties"), "Checkpoint file must be created");
        assertTrue(fileset.contains("0.page"), "One page file must be created");

        final Path pageFile = tempQueueFolder.resolve("0.page");
        verifyFile(pageFile, 9, rawContent -> {
            assertEquals(4, rawContent.getInt(), "First 4 bytes contains the length");
            assertEquals('A', rawContent.get());
            assertEquals('A', rawContent.get());
            assertEquals('A', rawContent.get());
            assertEquals('A', rawContent.get());
            assertEquals(0, rawContent.get());
        });
    }