@Override
    public ObjectNode encode(FlowInfo info, CodecContext context) {
        checkNotNull(info, "FlowInfo cannot be null");

        ObjectNode result = context.mapper().createObjectNode()
                .put(FLOW_TYPE, info.flowType())
                .put(DEVICE_ID, info.deviceId().toString())
                .put(INPUT_INTERFACE_ID, info.inputInterfaceId())
                .put(OUTPUT_INTERFACE_ID, info.outputInterfaceId())
                .put(SRC_IP, info.srcIp().address().toString())
                .put(SRC_IP_PREFIX_LEN, info.srcIp().prefixLength())
                .put(DST_IP, info.dstIp().address().toString())
                .put(DST_IP_PREFIX_LEN, info.dstIp().prefixLength())
                .put(SRC_PORT, info.srcPort().toString())
                .put(DST_PORT, info.dstPort().toString())
                .put(PROTOCOL, info.protocol())
                .put(SRC_MAC, info.srcMac().toString())
                .put(DST_MAC, info.dstMac().toString());

        if (info.vlanId() != null) {
            result.put(VLAN_ID, info.vlanId().toString());
        } else {
            result.put(VXLAN_ID, info.vxlanId());
        }

        ObjectNode statsInfoJson =
                context.codec(StatsInfo.class).encode(info.statsInfo(), context);

        result.put(STATS_INFO, statsInfoJson);

        return result;
    }
@Test
    public void testEncode() {
        StatsInfo statsInfo = new DefaultStatsInfo.DefaultBuilder()
                                    .withStartupTime(LONG_VALUE)
                                    .withFstPktArrTime(LONG_VALUE)
                                    .withLstPktOffset(INTEGER_VALUE)
                                    .withPrevAccBytes(LONG_VALUE)
                                    .withPrevAccPkts(INTEGER_VALUE)
                                    .withCurrAccBytes(LONG_VALUE)
                                    .withCurrAccPkts(INTEGER_VALUE)
                                    .withErrorPkts(SHORT_VALUE)
                                    .withDropPkts(SHORT_VALUE)
                                    .build();
        FlowInfo flowInfo = new DefaultFlowInfo.DefaultBuilder()
                                    .withFlowType((byte) FLOW_TYPE)
                                    .withDeviceId(DeviceId.deviceId(DEVICE_ID))
                                    .withInputInterfaceId(INPUT_INTERFACE_ID)
                                    .withOutputInterfaceId(OUTPUT_INTERFACE_ID)
                                    .withVlanId(VlanId.vlanId((short) VLAN_ID))
                                    .withSrcIp(IpPrefix.valueOf(
                                            IpAddress.valueOf(SRC_IP_ADDRESS), SRC_IP_PREFIX))
                                    .withDstIp(IpPrefix.valueOf(
                                            IpAddress.valueOf(DST_IP_ADDRESS), DST_IP_PREFIX))
                                    .withSrcPort(TpPort.tpPort(SRC_PORT))
                                    .withDstPort(TpPort.tpPort(DST_PORT))
                                    .withProtocol((byte) PROTOCOL)
                                    .withSrcMac(MacAddress.valueOf(SRC_MAC_ADDRESS))
                                    .withDstMac(MacAddress.valueOf(DST_MAC_ADDRESS))
                                    .withStatsInfo(statsInfo)
                                    .build();

        ObjectNode nodeJson = flowInfoCodec.encode(flowInfo, context);
        assertThat(nodeJson, matchesFlowInfo(flowInfo));

        FlowInfo flowInfoDecoded = flowInfoCodec.decode(nodeJson, context);
        new EqualsTester().addEqualityGroup(flowInfo, flowInfoDecoded).testEquals();
    }