ReactiveHashCommands doesn't properly encode/decode Object value

[Dieken]

Describe the bug

ReactiveHashCommands<String, String, Object> uses Marshaller, this class encodes String into plain string without quotes, not JSON string, but decodes with Object.class.

I have to use ReactiveHashCommands<String, String, Object[]> to bypass, any chance to make ReactiveHashCommands<String, String, Object> just works?

Quarkus version or git rev

3.1

[quarkus-bot]

/cc @cescoffier (redis), @gsmet (redis), @machi1990 (redis)

[Dieken]

Maybe Marshaller.encode() can judge on parametric type on value, not class on value, if that valueType isn't String/Integer/Double, then just use Vert.X JSON to encode the value.

[cescoffier]

We just merged the support for custom codec. Can you try with a custom codec?

[cescoffier]

Btw we plan to use Type instead of Class - the custom codec PR already did the switch. Later we will allow TypeReference.

[Dieken]

We just merged the support for custom codec. Can you try with a custom codec?

I checked the PR, I guess I can't use a custom codec because I want to store any object into Redis hash table:

@ApplicationScoped
public class ObjectCodec implements Codec {
    @Override
    public boolean canHandle(Type clazz) {
        return clazz.equals(Object.class);      // <--- this probably doesn't work
    }

...

[Dieken]

Btw we plan to use Type instead of Class - the custom codec PR already did the switch. Later we will allow TypeReference.

Yes, this is what I want, I can pass Type to ReactiveRedisDataSource.hash(someType) to help Marshaller realize it should always use JSON.

[cescoffier]

So, would passing a Type be enough for you?

[Dieken]

So, would passing a Type be enough for you?

I think so, make Marshaller capture the declared class in generic type, not runtime class when calling encode().

[cescoffier]

We just merged the support for custom codec. Can you try with a custom codec?

I checked the PR, I guess I can't use a custom codec because I want to store any object into Redis hash table:

@ApplicationScoped
public class ObjectCodec implements Codec {
    @Override
    public boolean canHandle(Type clazz) {
        return clazz.equals(Object.class);      // <--- this probably doesn't work
    }

...

return true should work.

[Dieken]

We just merged the support for custom codec. Can you try with a custom codec?

I checked the PR, I guess I can't use a custom codec because I want to store any object into Redis hash table:

@ApplicationScoped
public class ObjectCodec implements Codec {
    @Override
    public boolean canHandle(Type clazz) {
        return clazz.equals(Object.class);      // <--- this probably doesn't work
    }

...

return true should work.

Doesn't that mean there can be only one codec? I can't have another codec, such as PersonCodec in the PR.

[cescoffier]

Yes, then it will be up to the order of the custom codecs. I agree it could be better.

However, using type/type reference is going to take quite some time (I need to modify 300 classes)

[Dieken]

@cescoffier

Verified latest main branch(commit 2da3206), it still breaks:

quarkus/extensions/redis-client/runtime/src/main/java/io/quarkus/redis/runtime/datasource/Marshaller.java

Line 50 in ed8b789

if (o instanceof String) {

For ReactiveHashCommands<String, String, Object> hashCommands = redis.hash(new TypeReference<Object>() { }); and ReactiveHashCommands<String, String, Object> hashCommands = redis.hash(Object.class);, the Marshaller
always encodes String/Double/Integer/byte[] NOT as JSON (actually Double and Integer encodings are proper JSON by accident), but decode(Object.class) always as JSON, thus it still breaks.

I still have to bypass with ReactiveHashCommands<String, String, Object[]> hashCommands.

[Dieken]

decode() meets step not "step", thus throws exception:

io.vertx.core.json.DecodeException: 
Failed to decode:Unrecognized token 'step': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (io.netty.buffer.ByteBufInputStream); line: 1, column: 6]
        at io.quarkus.vertx.runtime.jackson.QuarkusJacksonJsonCodec.fromParser(QuarkusJacksonJsonCodec.java:128)
        at io.quarkus.vertx.runtime.jackson.QuarkusJacksonJsonCodec.fromBuffer(QuarkusJacksonJsonCodec.java:100)
        at io.vertx.core.json.Json.decodeValue(Json.java:119)
        at io.quarkus.redis.datasource.codecs.Codecs$JsonCodec.decode(Codecs.java:81)
        at io.quarkus.redis.runtime.datasource.Marshaller.decode(Marshaller.java:97)
        at io.quarkus.redis.runtime.datasource.Marshaller.decode(Marshaller.java:88)
        at io.quarkus.redis.runtime.datasource.Marshaller.decodeAsMap(Marshaller.java:122)
        at io.quarkus.redis.runtime.datasource.AbstractHashCommands.decodeMap(AbstractHashCommands.java:178)
        at io.smallrye.context.impl.wrappers.SlowContextualFunction.apply(SlowContextualFunction.java:21)
        at io.smallrye.mutiny.operators.uni.UniOnItemTransform$UniOnItemTransformProcessor.onItem(UniOnItemTransform.java:36)
        at io.smallrye.mutiny.vertx.AsyncResultUni.lambda$subscribe$1(AsyncResultUni.java:35)
        at io.smallrye.mutiny.vertx.DelegatingHandler.handle(DelegatingHandler.java:25)
        at io.quarkus.redis.runtime.client.ObservableRedis.lambda$send$1(ObservableRedis.java:54)
        at io.vertx.core.impl.future.FutureImpl$3.onSuccess(FutureImpl.java:141)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
        at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
        at io.vertx.core.impl.future.Composition$1.onSuccess(Composition.java:62)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
        at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
        at io.vertx.core.impl.future.Eventually$1.onSuccess(Eventually.java:44)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:60)
        at io.vertx.core.impl.future.SucceededFuture.addListener(SucceededFuture.java:88)
        at io.vertx.core.impl.future.Eventually.onSuccess(Eventually.java:41)
        at io.vertx.core.impl.future.FutureBase.lambda$emitSuccess$0(FutureBase.java:54)
        at io.vertx.core.impl.EventLoopContext.execute(EventLoopContext.java:86)
        at io.vertx.core.impl.DuplicatedContext.execute(DuplicatedContext.java:163)
        at io.vertx.core.impl.future.FutureBase.emitSuccess(FutureBase.java:51)
        at io.vertx.core.impl.future.FutureImpl.tryComplete(FutureImpl.java:211)
        at io.vertx.core.impl.future.PromiseImpl.tryComplete(PromiseImpl.java:23)
        at io.vertx.redis.client.impl.RedisStandaloneConnection.handle(RedisStandaloneConnection.java:409)
        at io.vertx.redis.client.impl.RESPParser.handleResponse(RESPParser.java:296)
        at io.vertx.redis.client.impl.RESPParser.handle(RESPParser.java:128)
        at io.vertx.redis.client.impl.RESPParser.handle(RESPParser.java:24)
        at io.vertx.core.net.impl.NetSocketImpl.lambda$new$1(NetSocketImpl.java:101)
        at io.vertx.core.streams.impl.InboundBuffer.handleEvent(InboundBuffer.java:255)
        at io.vertx.core.streams.impl.InboundBuffer.write(InboundBuffer.java:134)
        at io.vertx.core.net.impl.NetSocketImpl$DataMessageHandler.handle(NetSocketImpl.java:402)
        at io.vertx.core.impl.EventLoopContext.emit(EventLoopContext.java:55)
        at io.vertx.core.impl.DuplicatedContext.emit(DuplicatedContext.java:158)
        at io.vertx.core.net.impl.NetSocketImpl.handleMessage(NetSocketImpl.java:378)
        at io.vertx.core.net.impl.ConnectionBase.read(ConnectionBase.java:158)
        at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:153)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:442)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
        at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:412)
        at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1410)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:440)
        at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:420)
        at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:919)
        at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:166)
        at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:788)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:724)
        at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:650)
        at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:562)
        at io.netty.util.concurrent.SingleThreadEventExecutor$4.run(SingleThreadEventExecutor.java:997)
        at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74)
        at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)
        at java.base/java.lang.Thread.run(Thread.java:833)
Caused by: com.fasterxml.jackson.core.JsonParseException: Unrecognized token 'step': was expecting (JSON String, Number, Array, Object or token 'null', 'true' or 'false')
 at [Source: (io.netty.buffer.ByteBufInputStream); line: 1, column: 6]
        at com.fasterxml.jackson.core.JsonParser._constructError(JsonParser.java:2477)
        at com.fasterxml.jackson.core.base.ParserMinimalBase._reportError(ParserMinimalBase.java:760)
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._reportInvalidToken(UTF8StreamJsonParser.java:3699)
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._handleUnexpectedValue(UTF8StreamJsonParser.java:2787)
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser._nextTokenNotInObject(UTF8StreamJsonParser.java:908)
        at com.fasterxml.jackson.core.json.UTF8StreamJsonParser.nextToken(UTF8StreamJsonParser.java:794)
        at com.fasterxml.jackson.databind.ObjectMapper._initForReading(ObjectMapper.java:4912)
        at com.fasterxml.jackson.databind.ObjectMapper._readValue(ObjectMapper.java:4793)
        at com.fasterxml.jackson.databind.ObjectMapper.readValue(ObjectMapper.java:2974)
        at io.quarkus.vertx.runtime.jackson.QuarkusJacksonJsonCodec.fromParser(QuarkusJacksonJsonCodec.java:125)
        ... 57 more

[cescoffier]

You will have to regiester a custom codec, otherwise yes, it will encode the basic types using the expected RESP encoding.

[Dieken]

As #34554 (comment) described, I can’t use custom codec for Object.

[cescoffier]

If you do not have a known type, you must handle the encoding yourself (using the json mapper).
We cannot change how the basic RESP types are handled - this would break the interoperability with the other redis clients (The RESP protocol defines how these objects must be written)

[cescoffier]

BTW, for such kind of generic usage, I would even recommend using the low-level client and not the type-safe data source.

[Dieken]

@cescoffier Could this be backported to 3.2.x ? or when will 3.3.x be released? I decided to use ReactiveHashCommands<String, String, Var> and require custom codec for class Var.

// ReactiveHashCommands can't encode Object properly.
public class Var {

    private Object obj;

    public Var(Object obj) {
        this.obj = obj;
    }

    public Object get() {
        return obj;
    }
}

@ApplicationScoped
public class VarCodec implements Codec<Var> {

    private ObjectMapper mapper;

    public VarCodec(ObjectMapper mapper) {
        this.mapper = mapper.copy()
                .activateDefaultTyping(new VarTypeValidator(), DefaultTyping.EVERYTHING);
    }

    @Override
    public byte[] encode(Var item) {
        try {
            return mapper.writeValueAsBytes(item.get());
        } catch (JsonProcessingException ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public Var decode(byte[] item) {
        try {
            return new Var(mapper.readValue(item, Object.class));
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }
}

public class VarTypeValidator extends PolymorphicTypeValidator {

    @Override
    public Validity validateBaseType(MapperConfig<?> config, JavaType baseType) {
        return Validity.ALLOWED;
    }

    @Override
    public Validity validateSubClassName(MapperConfig<?> config, JavaType baseType, String subClassName) throws JsonMappingException {
        return Validity.ALLOWED;
    }

    @Override
    public Validity validateSubType(MapperConfig<?> config, JavaType baseType, JavaType subType) throws JsonMappingException {
        return Validity.ALLOWED;
    }
}

[Dieken]

Got a dirty way to save a little CPU and memory with custom codec in Quarkus main branch😄

Notice the decode() doesn't return type Var, so I don't have to unwrap Var into Object when deserialize,
just have to wrap Object into Var when serialize. The cost is I have to declare ReactiveHashCommands varsCommands = redis.hash(Var.class) without Java Generic.

@ApplicationScoped
public class VarCodec implements Codec {

    private ObjectMapper mapper;

    public VarCodec(ObjectMapper mapper) {
        this.mapper = mapper.copy()
                .activateDefaultTyping(new VarTypeValidator(), DefaultTyping.EVERYTHING);
    }

    @Override
    public byte[] encode(Object item) {
        try {
            return mapper.writeValueAsBytes(((Var) item).get());
        } catch (JsonProcessingException ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public Object decode(byte[] item) {
        try {
            return mapper.readValue(item, Object.class);
        } catch (IOException ex) {
            throw new RuntimeException(ex);
        }
    }

    @Override
    public boolean canHandle(Type clazz) {
        return Var.class.equals(clazz);
    }
}

[Dieken]

Could we have API like default <V> ReactiveHashCommands<String, String, V> hash(Class<V> typeOfValue, Codec<V> codec)?

I wouldn't like to use low level RedisAPI because I don't want to repeat logic in AbstractHashCommands.decodeMap() and Mashaller.decodeAsMap().

@cescoffier