public static NetFlowV5Packet parsePacket(ByteBuf bb) {
        final int readableBytes = bb.readableBytes();

        final NetFlowV5Header header = parseHeader(bb.slice(bb.readerIndex(), HEADER_LENGTH));
        final int packetLength = HEADER_LENGTH + header.count() * RECORD_LENGTH;
        if (header.count() <= 0 || readableBytes < packetLength) {
            throw new CorruptFlowPacketException("Insufficient data (expected: " + packetLength + " bytes, actual: " + readableBytes + " bytes)");
        }

        final ImmutableList.Builder<NetFlowV5Record> records = ImmutableList.builder();
        int offset = HEADER_LENGTH;
        for (int i = 0; i < header.count(); i++) {
            records.add(parseRecord(bb.slice(offset + bb.readerIndex(), RECORD_LENGTH)));
            offset += RECORD_LENGTH;
        }

        return NetFlowV5Packet.create(header, records.build(), offset);
    }
@Test
    public void pcap_pmacctd_NetFlowV5() throws Exception {
        final List<NetFlowV5Record> allRecords = new ArrayList<>();
        try (InputStream inputStream = Resources.getResource("netflow-data/pmacctd-netflow5.pcap").openStream()) {
            final Pcap pcap = Pcap.openStream(inputStream);
            pcap.loop(packet -> {
                        if (packet.hasProtocol(Protocol.UDP)) {
                            final UDPPacket udp = (UDPPacket) packet.getPacket(Protocol.UDP);
                            final ByteBuf byteBuf = Unpooled.wrappedBuffer(udp.getPayload().getArray());
                            final NetFlowV5Packet netFlowV5Packet = NetFlowV5Parser.parsePacket(byteBuf);
                            assertThat(netFlowV5Packet).isNotNull();
                            allRecords.addAll(netFlowV5Packet.records());
                        }
                        return true;
                    }
            );
        }
        assertThat(allRecords).hasSize(42);
    }