@Nonnull
    @Override
    public Result addChunk(ByteBuf buf, @Nullable SocketAddress remoteAddress) {
        if (!buf.isReadable(2)) {
            return new Result(null, false);
        }
        try {
            final IpfixParser.MessageDescription messageDescription = shallowParser.shallowParseMessage(buf);
            final long observationDomainId = messageDescription.getHeader().observationDomainId();
            addTemplateKeyInCache(remoteAddress, messageDescription, observationDomainId);
            // TODO handle options templates

            // collects all data records that are now ready to be sent
            final Set<ShallowDataSet> packetsToSendCollection = new HashSet<>();

            // the set of template records to include in the newly created message that is our "aggregate result"
            final Set<Integer> bufferedTemplateIdList = new HashSet<>();

            if (!messageDescription.declaredTemplateIds().isEmpty()) {
                // if we have new templates, look for buffered data records that we have all the templates for now
                final Set<Integer> knownTemplateIdsList = new HashSet<>();
                collectAllTemplateIds(remoteAddress, observationDomainId, knownTemplateIdsList);

                final Queue<ShallowDataSet> bufferedPackets = packetCache.getIfPresent(TemplateKey.idForExporter(remoteAddress, observationDomainId));
                handleBufferedPackets(packetsToSendCollection, bufferedTemplateIdList, knownTemplateIdsList, bufferedPackets);
            }
            boolean packetBuffered = false;

            // the list of template keys to return in the result ( TODO this copies all of the templates all the time :( )
            final Set<TemplateKey> templatesList = new HashSet<>(templateCache.asMap().keySet());

            bufferedTemplateIdList.addAll(messageDescription.referencedTemplateIds());
            LOG.debug("Finding the needed templates for the buffered and current packets");
            for (int templateId : bufferedTemplateIdList) {
                final TemplateKey templateKey = new TemplateKey(remoteAddress, observationDomainId, templateId);
                final Object template = templateCache.getIfPresent(templateKey);

                if (template == null) {
                    LOG.debug("Template is null, packet needs to be buffered until templates have been received.");
                    try {
                        final TemplateKey newTemplateKey = TemplateKey.idForExporter(remoteAddress, observationDomainId);
                        final Queue<ShallowDataSet> bufferedPackets = packetCache.get(newTemplateKey, ConcurrentLinkedQueue::new);
                        final byte[] bytes = ByteBufUtil.getBytes(buf);
                        bufferedPackets.addAll(messageDescription.dataSets());
                        packetBuffered = true;
                    } catch (ExecutionException ignored) {
                        // the loader cannot fail, it only creates a new queue
                    }
                } else {
                    LOG.debug("Template [{}] has been added to template list.", templateKey);
                    templatesList.add(templateKey);
                    packetsToSendCollection.addAll(messageDescription.dataSets());
                }
            }

            // if we have buffered this packet, don't try to process it now. we still need all the templates for it
            if (packetBuffered) {
                LOG.debug("Packet has been buffered and will not be processed now, returning result.");
                return new Result(null, true);
            }

            // if we didn't buffer anything but also didn't have anything queued that can be processed, don't proceed.
            if (packetsToSendCollection.isEmpty()) {
                LOG.debug("Packet has not been buffered and no packet is queued.");
                return new Result(null, true);
            }

            final IpfixJournal.RawIpfix.Builder journalBuilder = IpfixJournal.RawIpfix.newBuilder();
            buildJournalObject(packetsToSendCollection, templatesList, journalBuilder);
            final IpfixJournal.RawIpfix rawIpfix = journalBuilder.build();
            return getCompleteResult(rawIpfix);

        } catch (Exception e) {
            LOG.error("Unable to aggregate IPFIX message due to the following error ", e);
            return new Result(null, false);
        }
    }
@SuppressWarnings("unchecked")
    @Test
    public void ixFlowTest() throws IOException, URISyntaxException {
        final IpfixAggregator ipfixAggregator = new IpfixAggregator();
        final Map<String, Object> configMap = getIxiaConfigmap();
        final IpfixCodec codec = new IpfixCodec(new Configuration(configMap), ipfixAggregator, messageFactory);
        final List<Message> messages = new ArrayList<>();

        // ixflow.pcap contains 4 packets, the first has the data templates and option templates
        // followed by three data sets. two sets have subtemplateList data, the third has only empty lists for domain information
        try (InputStream stream = Resources.getResource("ixflow.pcap").openStream()) {
            final Pcap pcap = Pcap.openStream(stream);
            pcap.loop(packet -> {
                if (packet.hasProtocol(Protocol.UDP)) {
                    final UDPPacket udp = (UDPPacket) packet.getPacket(Protocol.UDP);
                    final InetSocketAddress source = new InetSocketAddress(udp.getParentPacket().getSourceIP(), udp.getSourcePort());
                    byte[] payload = new byte[udp.getPayload().getReadableBytes()];
                    udp.getPayload().getBytes(payload);
                    final ByteBuf buf = Unpooled.wrappedBuffer(payload);
                    final CodecAggregator.Result result = ipfixAggregator.addChunk(buf, source);
                    final ByteBuf ipfixRawBuf = result.getMessage();
                    if (ipfixRawBuf != null) {
                        byte[] bytes = new byte[ipfixRawBuf.readableBytes()];
                        ipfixRawBuf.getBytes(0, bytes);
                        messages.addAll(Objects.requireNonNull(codec.decodeMessages(new RawMessage(bytes))));
                    }
                }
                return true;
            });
        } catch (IOException e) {
            fail("Cannot process PCAP stream");
        }

        assertThat(messages).hasSize(3);
        assertThat(messages.get(0).getFields())
                .doesNotContainKey("httpSession")
                .containsEntry("dnsRecord_0_dnsIpv4Address", "1.2.0.2")
                .containsEntry("dnsRecord_0_dnsIpv6Address", "0:0:0:0:0:0:0:0")
                .containsEntry("dnsRecord_0_dnsName", "server-1020002.example.int.");
        assertThat(messages.get(1).getFields())
                .doesNotContainKey("httpSession")
                .containsEntry("dnsRecord_0_dnsIpv4Address", "1.2.14.73")
                .containsEntry("dnsRecord_0_dnsIpv6Address", "0:0:0:0:0:0:0:0")
                .containsEntry("dnsRecord_0_dnsName", "server-1020e49.example.int.");
        assertThat(messages.get(2).getFields())
                .doesNotContainKey("httpSession")
                .doesNotContainKey("dnsRecord");

    }