scatteringRead() — netty Function Reference

Source Code

transport-classes-epoll/src/main/java/io/netty/channel/epoll/EpollDatagramChannel.java lines 755–813

    private boolean scatteringRead(EpollRecvByteAllocatorHandle allocHandle, NativeDatagramPacketArray array,
            ByteBuf byteBuf, int datagramSize, int numDatagram) throws IOException {
        RecyclableArrayList datagramPackets = null;
        try {
            int offset = byteBuf.writerIndex();
            for (int i = 0; i < numDatagram;  i++, offset += datagramSize) {
                if (!array.addWritable(byteBuf, offset, datagramSize)) {
                    break;
                }
            }

            allocHandle.attemptedBytesRead(offset - byteBuf.writerIndex());

            NativeDatagramPacketArray.NativeDatagramPacket[] packets = array.packets();

            int received = socket.recvmmsg(packets, 0, array.count());
            if (received == 0) {
                allocHandle.lastBytesRead(-1);
                return false;
            }

            InetSocketAddress local = localAddress();

            // Set the writerIndex too the maximum number of bytes we might have read.
            int bytesReceived = received * datagramSize;
            byteBuf.writerIndex(byteBuf.writerIndex() + bytesReceived);

            if (received == 1) {
                // Single packet fast-path
                DatagramPacket packet = packets[0].newDatagramPacket(byteBuf, local);
                if (!(packet instanceof io.netty.channel.unix.SegmentedDatagramPacket)) {
                    processPacket(pipeline(), allocHandle, datagramSize, packet);
                    return true;
                }
            }
            // Its important that we process all received data out of the NativeDatagramPacketArray
            // before we call fireChannelRead(...). This is because the user may call flush()
            // in a channelRead(...) method and so may re-use the NativeDatagramPacketArray again.
            datagramPackets = RecyclableArrayList.newInstance();
            for (int i = 0; i < received; i++) {
                DatagramPacket packet = packets[i].newDatagramPacket(byteBuf, local);

                // We need to skip the maximum datagram size to ensure we have the readerIndex in the right position
                // for the next one.
                byteBuf.skipBytes(datagramSize);
                addDatagramPacketToOut(packet, datagramPackets);
            }
            // Ass we did use readRetainedSlice(...) before we should now release the byteBuf and null it out.
            byteBuf.release();
            byteBuf = null;

            processPacketList(pipeline(), allocHandle, bytesReceived, datagramPackets);
            datagramPackets.recycle();
            datagramPackets = null;
            return true;
        } finally {
            releaseAndRecycle(byteBuf, datagramPackets);
        }
    }