class RedisStateMapJsonSerDe implements RedisStateMapSerDe {

    @Override
    public RedisStateMap deserialize(final byte[] data) throws IOException {
        if (data == null || data.length == 0) {
            return null;
        }

        final RedisStateMap.Builder builder = new RedisStateMap.Builder();

        try (final JsonParser jsonParser = jsonFactory.createParser(data)) {
            final JsonNode rootNode = jsonParser.readValueAsTree();
            builder.version(rootNode.get(FIELD_VERSION).asLong());
            builder.encodingVersion(rootNode.get(FIELD_ENCODING).asInt());

            final JsonNode stateValuesNode = rootNode.get(FIELD_STATE_VALUES);
            stateValuesNode.fields().forEachRemaining(e -> builder.stateValue(e.getKey(), e.getValue().asText()));
        }

        return builder.build();
    }

}