Support Top Level Value/Key Codecs

[ctoestreich]

The default client and factory use a simple codec <String, String>. It is pretty easy to override this, however it isn't documented and the ability to specify a key/value codec could be simplified.

I had to do the following to get a custom protobuf session codec working where the UserSession is a proto we are using. Seems like overkill to have to redefine the entire class to achieve this result.

Also there isn't examples or documentation on the https://micronaut-projects.github.io/micronaut-redis/latest/guide/ site for serializing objects that do not use the session store/jackson. Perhaps the jackson/other serialization could simply be added as top level option like

redis:
    valueSerializer: io.micronaut.jackson.serialize.JacksonObjectSerializer
    keySerializer: io.micronaut.jackson.serialize.JacksonObjectSerializer #or string etc

package xxxxxxxxxxxx;

import xxxxxxxxxxxx.UserSession;
import io.lettuce.core.RedisClient;
import io.lettuce.core.RedisURI;
import io.lettuce.core.api.StatefulRedisConnection;
import io.lettuce.core.codec.RedisCodec;
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
import io.micronaut.configuration.lettuce.AbstractRedisConfiguration;
import io.micronaut.configuration.lettuce.DefaultRedisClientFactory;
import io.micronaut.configuration.lettuce.DefaultRedisConfiguration;
import io.micronaut.context.annotation.Bean;
import io.micronaut.context.annotation.Factory;
import io.micronaut.context.annotation.Primary;
import io.micronaut.context.annotation.Replaces;
import io.micronaut.context.annotation.Requires;

import javax.inject.Singleton;
import java.nio.ByteBuffer;
import java.nio.charset.Charset;
import java.util.Optional;

@Requires(beans = DefaultRedisConfiguration.class)
@Singleton
@Factory
@Replaces(DefaultRedisClientFactory.class)
public class UserSessionRedisClientFactory {

    @Bean(preDestroy = "shutdown")
    @Singleton
    @Primary
    public RedisClient redisClient(AbstractRedisConfiguration config) {
        Optional<RedisURI> uri = config.getUri();
        return uri.map(RedisClient::create)
                .orElseGet(() -> RedisClient.create(config));
    }

    /**
     * Creates the {@link StatefulRedisConnection} from the {@link RedisClient}.
     *
     * @param redisClient The {@link RedisClient}
     * @return The {@link StatefulRedisConnection}
     */
    @Bean(preDestroy = "close")
    @Singleton
    @Primary
    public StatefulRedisConnection<String, UserSession> redisConnection(RedisClient redisClient) {
        return redisClient.connect(new UserSessionCodec());
    }

    /**
     * Creates the {@link StatefulRedisPubSubConnection} from the {@link RedisClient}.
     *
     * @param redisClient The {@link RedisClient}
     * @return The {@link StatefulRedisPubSubConnection}
     */
    @Bean(preDestroy = "close")
    @Singleton
    public StatefulRedisPubSubConnection<String, String> redisPubSubConnection(RedisClient redisClient) {
        return redisClient.connectPubSub();
    }

    public class UserSessionCodec implements RedisCodec<String, UserSession> {
        private Charset charset = Charset.forName("UTF-8");

        @Override
        public String decodeKey(ByteBuffer bytes) {
            return charset.decode(bytes).toString();
        }

        @Override
        public UserSession decodeValue(ByteBuffer bytes) {
            try {
                byte[] array = new byte[bytes.remaining()];
                bytes.get(array);
                return UserSession.parseFrom(array);
            } catch (Exception e) {
                return null;
            }
        }

        @Override
        public ByteBuffer encodeKey(String key) {
            return charset.encode(key);
        }

        @Override
        public ByteBuffer encodeValue(UserSession value) {
            return ByteBuffer.wrap(value.toByteArray());
        }
    }
}