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 insertDataTriggerCreationOfNewPageFile() throws QueueException {
        final QueuePool queuePool = QueuePool.loadQueues(tempQueueFolder, PAGE_SIZE, SEGMENT_SIZE);
        final Queue queue = queuePool.getOrCreate("test");

        // one page is 64 MB so the loop count to fill it is 64 * 1024
        // 4 bytes are left for length so that each time are inserted 1024 bytes, 4 header and 1020 payload
        ByteBuffer payload = ByteBuffer.wrap(generatePayload(1024 - LENGTH_HEADER_SIZE));
        for (int i = 0; i < 64; i++) {
            writeMessages(queue, payload, 1024);
        }

        // check the 2 files are created
        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");

        // Exercise
        // some data to force create a new page
        final ByteBuffer crossingPayload = ByteBuffer.wrap(generatePayload(10, (byte) 'B'));
        queue.enqueue(crossingPayload);

        // Verify
        fileset = new HashSet<>(Arrays.asList(tempQueueFolder.toFile().list()));
        assertEquals(3, fileset.size());
        assertTrue(fileset.contains("checkpoint.properties"), "Checkpoint file must be created");
        assertTrue(fileset.contains("0.page"), "First page file must be created");
        assertTrue(fileset.contains("1.page"), "Second page file must be created");
    }