class RedisStateMapJsonSerDe implements RedisStateMapSerDe {

    @Override
    public byte[] serialize(final RedisStateMap stateMap) throws IOException {
        if (stateMap == null) {
            return null;
        }

        try (final ByteArrayOutputStream out = new ByteArrayOutputStream()) {
            final JsonGenerator jsonGenerator = jsonFactory.createGenerator(out);
            jsonGenerator.writeStartObject();
            jsonGenerator.writeNumberField(FIELD_VERSION, stateMap.getVersion());
            jsonGenerator.writeNumberField(FIELD_ENCODING, stateMap.getEncodingVersion());

            jsonGenerator.writeObjectFieldStart(FIELD_STATE_VALUES);
            for (Map.Entry<String,String> entry : stateMap.toMap().entrySet()) {
                jsonGenerator.writeStringField(entry.getKey(), entry.getValue());
            }
            jsonGenerator.writeEndObject();

            jsonGenerator.writeEndObject();
            jsonGenerator.flush();

            return out.toByteArray();
        }
    }

}