@Nullable
    @Override
    public Message decode(@Nonnull RawMessage rawMessage) {
        String s = new String(rawMessage.getPayload(), StandardCharsets.UTF_8);
        LOG.trace("Received raw message: {}", s);

        String timezoneID = configuration.getString(CK_TIMEZONE);
        // previously existing PA inputs after updating will not have a Time Zone configured, default to UTC
        DateTimeZone timezone = timezoneID != null ? DateTimeZone.forID(timezoneID) : DateTimeZone.UTC;
        LOG.trace("Configured time zone: {}", timezone);
        PaloAltoMessageBase p = parser.parse(s, timezone);

        // Return when error occurs parsing syslog header.
        if (p == null) {
            return null;
        }

        Message message = messageFactory.createMessage(p.payload(), p.source(), p.timestamp());

        switch (p.panType()) {
            case "THREAT":
                final PaloAltoTypeParser parserThreat = new PaloAltoTypeParser(templates.getThreatMessageTemplate());
                message.addFields(parserThreat.parseFields(p.fields(), timezone));
                break;
            case "SYSTEM":
                final PaloAltoTypeParser parserSystem = new PaloAltoTypeParser(templates.getSystemMessageTemplate());
                message.addFields(parserSystem.parseFields(p.fields(), timezone));
                break;
            case "TRAFFIC":
                final PaloAltoTypeParser parserTraffic = new PaloAltoTypeParser(templates.getTrafficMessageTemplate());
                message.addFields(parserTraffic.parseFields(p.fields(), timezone));
                break;
            default:
                LOG.error("Unsupported PAN type [{}]. Not adding any parsed fields.", p.panType());
        }

        LOG.trace("Successfully processed [{}] message with [{}] fields.", p.panType(), message.getFieldCount());

        return message;
    }
@Test
    public void valuesTest() {

        // Test System message results
        PaloAltoCodec codec = new PaloAltoCodec(Configuration.EMPTY_CONFIGURATION, messageFactory);
        Message message = codec.decode(new RawMessage(PANORAMA_SYSTEM_MESSAGE.getBytes(StandardCharsets.UTF_8)));
        assertEquals("SYSTEM", message.getField("type"));
        assertEquals(message.getField("module"), "general");

        // Test quoted value with embedded commas.
        assertEquals(message.getField("description"), "Deviating device: Prod--2, Serial: 453524335, Object: N/A, Metric: mp-cpu, Value: 34");
        assertEquals(message.getField("serial_number"), "000710000506");
        assertEquals(message.getField("source"), "Panorama-1");
        assertEquals(message.getField("message"), "1,2018/09/19 11:50:35,000710000506,SYSTEM,general,0,2018/09/19 11:50:35,,general,,0,0,general,informational,\"Deviating device: Prod--2, Serial: 453524335, Object: N/A, Metric: mp-cpu, Value: 34\",1163103,0x0,0,0,0,0,,Panorama-1");
        assertEquals(message.getField("severity"), "informational");
        assertEquals(message.getField("generated_time"), "2018/09/19 11:50:35");
        assertEquals(message.getField("event_id"), "general");
        assertEquals(message.getField("device_name"), "Panorama-1");
        assertEquals(message.getField("content_threat_type"), "general");
        assertEquals(message.getField("virtual_system_name"), null);
        assertEquals(0, ((DateTime) message.getField("timestamp")).compareTo(new DateTime("2018-09-19T11:50:35.000-05:00", DateTimeZone.UTC)));

        // Test Traffic message results
        message = codec.decode(new RawMessage(PANORAMA_TRAFFIC_MESSAGE.getBytes(StandardCharsets.UTF_8)));
        assertEquals(message.getField("bytes_received"), 140L);
        assertEquals(message.getField("source"), "Panorama--2");
        assertEquals(message.getField("repeat_count"), 1L);
        assertEquals(message.getField("receive_time"), "2018/09/19 11:50:32");
        assertEquals(message.getField("outbound_interface"), "ethernet1/1");
        assertEquals(message.getField("packets"), 6L);
        assertEquals(message.getField("dest_location"), "10.20.30.40-10.20.30.40");
        assertEquals(message.getField("src_addr"), "10.20.30.40");
        assertEquals(message.getField("generated_time"), "2018/09/19 11:50:32");
        assertEquals(message.getField("protocol"), "tcp");
        assertEquals(message.getField("threat_content_type"), "end");
        assertEquals(message.getField("packets_sent"), 4L);
        assertEquals(message.getField("packets_received"), 2L);
        assertEquals(message.getField("action"), "allow");
        assertEquals(message.getField("virtual_system"), "vsys1");
        assertEquals(message.getField("dest_port"), 443L);
        assertEquals(((DateTime) message.getField("timestamp")).compareTo(new DateTime("2018-09-19T11:50:32.000-05:00", DateTimeZone.UTC)), 0);
        assertEquals(message.getField("rule_name"), "HTTPS-strict");
        assertEquals(message.getField("nat_src_addr"), "10.20.30.40");
        assertEquals(message.getField("session_id"), 205742L);
        assertEquals(message.getField("serial_number"), "453524335");
        assertEquals(message.getField("message"), "1,2018/09/19 11:50:32,453524335,TRAFFIC,end,2049,2018/09/19 11:50:32,10.20.30.40,10.20.30.40,10.20.30.40,10.20.30.40,HTTPS-strict,,,incomplete,vsys1,Public,Public,ethernet1/1,ethernet1/1,ALK Logging,2018/09/19 11:50:32,205742,1,64575,443,41304,443,0x400070,tcp,allow,412,272,140,6,2018/09/19 11:50:15,0,any,0,54196730,0x8000000000000000,10.20.30.40-10.20.30.40,10.20.30.40-10.20.30.40,0,4,2,tcp-fin,13,16,0,0,,Prod--2,from-policy,,,0,,0,,N/A,0,0,0,0");
        assertEquals(message.getField("bytes_sent"), 272L);
        assertEquals(message.getField("dest_zone"), "Public");
        assertEquals(message.getField("nat_src_port"), 41304L);
        assertEquals(message.getField("src_port"), 64575L);
        assertEquals(message.getField("src_location"), "10.20.30.40-10.20.30.40");
        assertEquals(message.getField("log_action"), "ALK Logging");
        assertEquals(message.getField("inbound_interface"), "ethernet1/1");
        assertEquals(message.getField("application"), "incomplete");
        assertEquals(message.getField("src_zone"), "Public");
        assertEquals(message.getField("bytes"), 412L);
        assertEquals(message.getField("dest_addr"), "10.20.30.40");
        assertEquals(message.getField("type"), "TRAFFIC");
        assertEquals(message.getField("nat_dest_addr"), "10.20.30.40");
        assertEquals(message.getField("category"), "any");
        assertEquals(message.getField("nat_dest_port"), 443L);
    }