@Nullable
    @Override
    public Message decode(@Nonnull final RawMessage rawMessage) {
        final GELFMessage gelfMessage = new GELFMessage(rawMessage.getPayload(), rawMessage.getRemoteAddress());
        final String json = gelfMessage.getJSON(decompressSizeLimit, charset);

        final JsonNode node;

        try {
            node = objectMapper.readTree(json);
            if (node == null) {
                throw new IOException("null result");
            }
        } catch (final Exception e) {
            log.error("Could not parse JSON, first 400 characters: " +
                    StringUtils.abbreviate(json, 403), e);
            throw new IllegalStateException("JSON is null/could not be parsed (invalid JSON)", e);
        }

        try {
            validateGELFMessage(node, rawMessage.getId(), rawMessage.getRemoteAddress());
        } catch (IllegalArgumentException e) {
            log.trace("Invalid GELF message <{}>", node);
            throw e;
        }

        // Timestamp.
        final double messageTimestamp = timestampValue(node);
        final DateTime timestamp;
        if (messageTimestamp <= 0) {
            timestamp = rawMessage.getTimestamp();
        } else {
            // we treat this as a unix timestamp
            timestamp = Tools.dateTimeFromDouble(messageTimestamp);
        }

        final Message message = messageFactory.createMessage(
                stringValue(node, "short_message"),
                stringValue(node, "host"),
                timestamp
        );

        message.addField(Message.FIELD_FULL_MESSAGE, stringValue(node, "full_message"));

        final String file = stringValue(node, "file");

        if (file != null && !file.isEmpty()) {
            message.addField("file", file);
        }

        final long line = longValue(node, "line");
        if (line > -1) {
            message.addField("line", line);
        }

        // Level is set by server if not specified by client.
        final int level = intValue(node, "level");
        if (level > -1) {
            message.addField("level", level);
        }

        // Facility is set by server if not specified by client.
        final String facility = stringValue(node, "facility");
        if (facility != null && !facility.isEmpty()) {
            message.addField("facility", facility);
        }

        // Add additional data if there is some.
        final Iterator<Map.Entry<String, JsonNode>> fields = node.fields();

        while (fields.hasNext()) {
            final Map.Entry<String, JsonNode> entry = fields.next();

            String key = entry.getKey();
            // Do not index useless GELF "version" field.
            if ("version".equals(key)) {
                continue;
            }

            // Don't include GELF syntax underscore in message field key.
            if (key.startsWith("_") && key.length() > 1) {
                key = key.substring(1);
            }

            // We already set short_message and host as message and source. Do not add as fields again.
            if ("short_message".equals(key) || "host".equals(key)) {
                continue;
            }

            // Skip standard or already set fields.
            if (message.getField(key) != null || Message.RESERVED_FIELDS.contains(key) && !Message.RESERVED_SETTABLE_FIELDS.contains(key)) {
                continue;
            }

            // Convert JSON containers to Strings, and pick a suitable number representation.
            final JsonNode value = entry.getValue();

            final Object fieldValue;
            if (value.isContainerNode()) {
                fieldValue = value.toString();
            } else if (value.isFloatingPointNumber()) {
                fieldValue = value.asDouble();
            } else if (value.isIntegralNumber()) {
                fieldValue = value.asLong();
            } else if (value.isNull()) {
                log.debug("Field [{}] is NULL. Skipping.", key);
                continue;
            } else if (value.isTextual()) {
                fieldValue = value.asText();
            } else {
                log.debug("Field [{}] has unknown value type. Skipping.", key);
                continue;
            }

            message.addField(key, fieldValue);
        }

        return message;
    }
@Test
    public void decodeFailsWithBlankMessage() throws Exception {
        final String json = "{"
                + "\"version\": \"1.1\","
                + "\"host\": \"example.org\","
                + "\"message\": \"     \""
                + "}";

        final RawMessage rawMessage = new RawMessage(json.getBytes(StandardCharsets.UTF_8));

        assertThatIllegalArgumentException().isThrownBy(() -> codec.decode(rawMessage))
                .withNoCause()
                .withMessageMatching("GELF message <[0-9a-f-]+> has empty mandatory \"message\" field.");
    }