vert-x3 / vertx-redis-client

Redis client for Vert.x

Vert.x Redis Client

The Vert.x Redis client provides an asynchronous API to interact with a Redis data-structure server.

Documentation:

The documentation can be read from the official vert.x Documentation.

vertx-redis-client's Issues

ClassCastException, while using spopMany

Version

• vert.x core:3.2.1
• vert.x redis client:3.2.1

    redisClient.spopMany("FOO", 3, ar -> {
      if (ar.succeeded()) {
        System.out.println(ar.result());
      } else {
        System.out.println(ar.cause());
      }
    });

the result of SPOP key [count] should be JsonArray
exception:

java.lang.ClassCastException: io.vertx.core.json.JsonArray cannot be cast to java.lang.String

maven dependency

It would be nice to see in docs correct maven dependency:

<dependency>
    <groupId>io.vertx</groupId>
    <artifactId>vertx-redis-client</artifactId>
    <version>3.0.0</version>
</dependency>

It took me a while to find it.

make client less opionated

Currently the client will behind the scenes enqueue request until it manages to connect to the server, it will try to reconnect on error and hold messages that have not been processed and attempt to handle them once a connection is available. It will also switch from pub/sub mode to normal mode and establish 2 connections so regular calls do not interfere with each other.

Although this is quite nice for people who just want it to work, for the case where one wants total control of what is going on this is bad decision.

I'd propose to create a different client class that is just purely simple. Establish a connection and has several handlers, for exception, error, etc, and sends commands.

setWithOptions shall return a result of String type

setWithOptions shall return a result of String type, install of Void. Otherwise it is no way to know the exact result when NX option is used.
SET command return string OK or (nil)

Redis Sentinel support

Hi there, I was going through the source of RedisConnection and there's nothing in there to suggest Redis Sentinel is supported. Would you mind confirming?

Thanks!

Weird problem when sending request to a basic Vert.x server with a basic Redis client.

I am currently working on a project where I need to receive an id number through a requests and then look if it is stored in my Redis database.

My project works well for one request or two, but when testing it with Jmeter and sending 3000 requests I noticed a weird problem.

I made a way more basic project and I published it to github to understand what was going on.

There is actually two problems :

• The first one being that sometimes my project won't respond to Jmeter after like responding 200 requests or so... Jmeter keeps waiting for the reply.
• The other one is that sometimes I get this exception, it has the same effect on Jmeter than the first one.

io.vertx.core.impl.ContextImpl
SEVERE: Unhandled exception
java.lang.RuntimeException: Received a non pub/sub message without reply handler waiting:io.vertx.redis.impl.Reply@6a3c7ffb
at io.vertx.redis.impl.RedisConnection.handleReply(RedisConnection.java:92)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:154)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:6)
at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:309)
at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$27(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler$$Lambda$47/140297772.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.lambda$wrapTask$15(ContextImpl.java:312)
at io.vertx.core.impl.ContextImpl$$Lambda$6/60559178.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:217)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:31)
at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:124)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:111)
at java.lang.Thread.run(Thread.java:745)

The error really seems to be coming from my code since when I try to respond the request instantly without even calling my Redis client I manage to go through the whole 3000 requests without any problem.
I am using Vert.x 3.0.

blpop produces connection closed error when timeout has expired

Version

• vert.x core: 3.4.1
• vert.x redis client: 3.4.1
• using dockerhub redis: https://hub.docker.com/_/redis/
  • started with: docker run --name redis-container -p 6379:6379 -d redis redis-server --appendonly yes

Context

I encountered an exception which looks suspicious while issuing blpop

Do you have a reproducer?

no

Steps to reproduce

1. Assume a redis list with some key name
2. Use the vertx client to issue blpop request for that key name using a non-zero timeout argument
3. Do not add any elements to the list with that key name
4. When the blocking period has expired, the client produces a connection closed exception instead of a non-error response

Extra

• redis documentation indicates that a nil multi-bulk response should come back, so I'm expecting a non-error response with an empty JsonArray. I think it's important to be able to distinguish between true connection failures and expected blpop behavior.

bad behavior when running on high load

reproducer: https://github.com/exanj/testRedis

Redis pipeline

Hello!
Redis supports commands sent in a pipeline. This reduces network traffic as well as turns some commands into a sort of commit block. Is there any plans to implement a pipeline?

Selecting Database works for writing a value but not getting it back.

When So I wrote this test

According to the documentation selecting a database using select changes it for the current connection

  /**
   * Change the selected database for the current connection
   *
   * @param dbindex Index identifying the new active database
   * @param handler Handler for the result of this call.
   * @since 1.0.0
   * group: connection
   */
  @Fluent
  RedisClient select(int dbindex, Handler<AsyncResult<String>> handler);

But, when writing then reading, the value is not retrieved.
It works fine if I comment out the redis.select part.

import io.vertx.core.AsyncResult;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.ext.unit.TestContext;
import io.vertx.ext.unit.junit.RunTestOnContext;
import io.vertx.ext.unit.junit.VertxUnitRunner;
import io.vertx.redis.RedisClient;
import io.vertx.redis.RedisOptions;

import java.io.IOException;

import org.junit.AfterClass;
import org.junit.BeforeClass;
import org.junit.ClassRule;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

@RunWith(VertxUnitRunner.class)
public class RedisClientTest {
  private static final Logger LOG = LoggerFactory.getLogger(RedisClientTest.class);

  @ClassRule
  public static RunTestOnContext contextRule = new RunTestOnContext();
  private static RedisClient redis;

  private final static String KEY1 = "TheFirstKey";
  private static final int DB_NUM = 1;

  @BeforeClass
  public static void initializeRedis(TestContext context) throws InterruptedException, IOException {
    // Create the RMQ service and confirm that the connection was created
    Vertx vertx = contextRule.vertx();
    RedisOptions config = new RedisOptions().setHost("127.0.0.1").setTcpKeepAlive(true);
    redis = RedisClient.create(vertx, config);
    redis.select(DB_NUM, new Handler<AsyncResult<String>>() {
      @Override
      public void handle(AsyncResult<String> event) {
        LOG.info(event.result());
        if (event.succeeded())
          LOG.info("Redis switched to DB: {}",DB_NUM );
        if (event.failed()) {
          LOG.error("Failed to select DB");
          throw new RuntimeException("Failed to select DB");
        }
      }
    });

  }

  @AfterClass
  public static void stopRedis(TestContext context) throws InterruptedException, IOException {
    redis.del(KEY1, r -> {
        redis.close(event -> {
          if (event.succeeded())
            LOG.info("Redis connection closed");
          if (event.failed()) {
            LOG.error("Failed to close connection to Redis");
            throw new RuntimeException("Failed to close connection to Redis");
          }
        });
      });
  }

  @Test
  public void test(TestContext context) throws Throwable {
    String message = "this is a test";

    // Write and read a message to/from Redis
    redis.set(KEY1, message, r -> {
      if (r.succeeded()) {
        LOG.info("{} wrote to Redis with value: {}",KEY1, message);
        redis.get(KEY1, r2 -> {
          if (r2.failed())
            LOG.error("Read failed ", r2.cause());
          else{
            LOG.info("{} got from Redis with value: {}",KEY1, r2.result());
            context.assertEquals(message, r2.result());
          }
        });
      }
    });
  }

}

Am I doing something wrong ?

support redis cluster?

Hello, does the vertx-redis support redis cluster ? I could not found any thing about it. If support, could you give me a demo please.

can't continue subscribe while reconnected

Version

• vert.x core:3.3.3
• vert.x redis client:3.3.3

Context

can't continue subscribe while reconnected, or how to handle connection's reconnect event?

About vertx-redis-client, the Pub/Sub mode

Version

• vert.x core:3.2.1
• vert.x redis client:3.2.1
• Jedis 2.7.3
• MSGPACK 0.6.11

Context

I have a redis server for pub/sub.

Jedis is used to publish byte[] message to the redis channel,
and the byte array is produced by MSGPACK .

When I received the published msg from the eventbus, I cannot decode the msg with MSGPACK .

After debugging, I think the problem encounted at class:
io.vertx.redis.impl.AbstractRedisClient,
"case SUBSCRIBE: " of the send() method ,
the line:
"
message.put("message", replyData[2].asType(String.class, encoding));
"

After vertx-redis-client "asType" the Buffer to String , I cannot convert the string to the byte array that can be decoded by MSGPACK.

Do you have a reproducer?

Steps to reproduce

1. Create a test pojo class, for example :

public class TestPojo {

private String[] aa;
private int bb;
private String cc;

...omit the getter, setter
}

1. Use msgpack encode the obj to byte array:

   MessagePack packer = new MessagePack();
   packer.register(TestPojo.class);
   TestPojo to = new TestPojo();
   to.setAa(new String[]{"x1", "y2"});
   to.setBb(34);
   to.setCc("The Test Sentence!");
   
   byte[] msg = null;
   try {
       msg = packer.write(to);
   } catch (IOException e) {
       e.printStackTrace();
   }
   

2. Use the jedis client to publish the (byte[]) message:

   String channel = "test-channel";
   JedisPool pool = new JedisPool(new JedisPoolConfig(),  <redis_server_ip>,  <redis_server_port>);
   Jedis j = pool.getResource();
   j.publish(channel.getBytes("utf-8"), msg);
   

3. Use vertx-redis subscribe the channel, and consume the msg from eventbus :

   vertx.eventBus().<JsonObject>consumer("io.vertx.redis" + channel , ar -> {
       JsonObject o = ar.body();
       String msg = o.getJsonObject("value").getString("message");
   
       MessagePack MSGPACK = new MessagePack();
       MSGPACK.register(TestPojo.class);
       try {
           TestPojo testObj = MSGPACK.read(msg.getBytes("utf-8"), TestPojo.class);
           System.out.println("---consumer, testObj:" + testObj.toString());
       } catch (IOException e) {
           e.printStackTrace();
       }
   
   });
   

4. Run the vertx program first, then publish a msg to redis, you will get a msgpack decode exception
   "org.msgpack.MessageTypeException".

Expand binary support for Redis

Version

• vert.x core: 3.4.2
• vert.x redis client: 3.4.2

Context

We are using Redis to store constantly incoming binary data, storing it to file is not an option since there might be race conditions and information will be lost. The only append function Redis currently have is for String. Since you have setBinary function I figured I can get the current data in form of Buffer using getBinary function, append to the buffer and use setBinary, but to avoid race condition I must use transaction.multi, and thats where the biggest issue is, ## getBinary in RedisClient class returns a Buffer, but getBinary in RedisTransaction class returns a String. Converting byte [] to String ruins the data so its not an option.

Kryo serialization

Any plan on using Kryo serializer.
It's one of the fastest.
https://github.com/eishay/jvm-serializers/wiki
https://code.google.com/p/kryo/

And unlike protobuff it doesn't need any pre configuration.

Operation mgetBinary

Version

• vert.x core: 3.4.2
• vert.x redis client: 3.4.2

Context

The redis-operation "mget" accepts many keys and can handle binary data. I am missing a "get binary data for multiple keys"-operation in vertx-redis-client (for performance reasons, I don't want to have a loop with 'getBinary'). Is it out of scope to implement that? Or is there a workaround with some encoding magic?

Do you have a reproducer?

This is related to #40. I created a test with mget that fails for binary data. An operation "mgetManyBinary" that makes this test green would be great.
Add this to RedisClientTestBase.java:

@Test
public void testMgetBinary() {
final String mykey = makeKey();
final byte[] binary_data = {0,1,127,-1,-128};
redis.setBinary(mykey, Buffer.buffer(binary_data), reply0 ->{
assertTrue(reply0.succeeded());
redis.mgetMany(toList(mykey),reply1 ->{
assertTrue(reply1.succeeded());
assertArrayEquals(binary_data,reply1.result().getString(0).getBytes());
testComplete();
});
});
await();
}

RedisConnection state corrupted by connection RST

The RedisConnection update its state when it handles an exception because buffers could still be in the NetSocket and delivered which corrupt the state of the RedisConnection.

Change it to only close the NetSocket and perform this state update in the NetSocket#closeHandler when we are sure no buffer event will be delivered.

Incorrect types on RedisTransaction

List operations (lrange, lrem, etc...) receive Handler<AsyncResult<String>> instead of Handler<AsyncResult<JsonArray>> and raises a Cast exception when performing:

RedisTransaction transaction = client.transaction();
transaction.lrange("foo", 0, -1, event -> {
    String value = event.result(); // this line throws cast exception
});

redis pub/sub documentation error

in pub/sub mode, currently the channel prefix in the documentation is "redis.sub.", which is not working and should be "io.vertx.redis." according to repo test code.

vertx: 3.2.1
redis:3.2

Investigate redis push as ReadStream

Version

• vert.x core:
• vert.x redis client:

Context

I encountered an exception which looks suspicious while ...

Do you have a reproducer?

• Link to github project/gist

Steps to reproduce

1. ...
2. ...
3. ...
4. ...

Extra

• Anything that can be relevant

master slave configuration / amazon elasticache

How do you configure vertx-redis-client for master slave configuration / amazon elasticache?

RedisConnection must always uses an EventLoop thread

Currently the RedisConnection may unsafely share its state when:

• executed from an executeBlocking lambda
• used from a worker context

make sure that in these case we uses the correct thread.

exec returns array not void

see: https://groups.google.com/forum/#!topic/vertx/bdZ1F4PbcdQ

Performance issues when huge results / command args

Version

• vert.x core: 3.3.3
• vert.x redis client: 3.3.3

Context

800000 keys in Redis. Single core. Single Verticle instance. Single HTTP request.
Keys are :  25.1234567 (two numbers separated by a dot)

First use case (for the sake of the example I'm omitting the async part) :
JsonArray theKeys = redis.scan("25.*") => returns the 800000 keys (in real life there'll be more than 800000... under 26.1234567 for instance).
Then JsonArray theValues = mgetmany(theKeys)
I tried with both vertx-redis-client and jedis and there's a huge difference between both.

Second use case :

(Originally to see if scanwas the issue)

I added every "second part of the key" in a sorted set with this key as score (and fake values, unique though)
Thus, the "full listing" becomes :
 JsonArray items = redis.zrangebyscore("thekeys", 0, 1000000, ...)
And then, the same mgetMany as as in the first case scenario.

I measured the time of the zrangebyscore between jedis and vertx-redis and the results are stunning :

• 3s (which is huge...) for Jedis
• 23s for vertx-redis-client

Do you have a reproducer?

I'm in the process of creating one, just wanted to warn you asap.

Steps to reproduce

1. Insert a million values in a Redis sorted set
2. Try to zrangebyscore the whole sorted set with vertx-redis
3. Use the resulting JsonArray to perform another Redis operation
4. In the meantime, watch slowlog on the Redis server. You'll see Redis doesn't see "such big durations"

Extra

• I was digging through slowlog when I realized Redis wasn't seeing long durations at all. So it's all on Vert.x marshalling / unmarshalling part.
• I'll try to create a full reproducer, performing the inserts, then "benchmarking" jedis vs. vertx-redis in both cases mentionned above
• I think JMH should be involved in the process to see if JSON marshalling is the issue ?
• I opened this issue just for you to be informed about my experiments

Hope we can investigate together.

Scan return type

scan
zscan
sscan

Should return type JsonArray and not Void.

Same issue as with exec call before.

Redis client reconnect fails without AUTH

@pmlopes

My redis server has AUTh required. So when I initialize my redis client, I have to manually call AUTH command after creating the redis client. Now I am seeing my redis-client losing connection to my redis server - not sure why yet - but reconnecting to the redis server does not call AUTH command. Therefore, queued commands won't be able to run after trying reconnecting with the exception:

NOAUTH Authentication required.

How should we reconnect to redis server requiring AUTH? I'd think we should call AUTH commands in the connect function in the RedisConnection class. Thoughts?

Thanks.

hgetall bug?

i use hgetall get hash data ,sometime will bring this error
error msg:
'''
java.lang.ArrayIndexOutOfBoundsException: 66
at io.netty.buffer.UnpooledHeapByteBuf._getByte(UnpooledHeapByteBuf.java:292)
at io.netty.buffer.UnpooledHeapByteBuf.getByte(UnpooledHeapByteBuf.java:287)
at io.netty.buffer.WrappedByteBuf.getByte(WrappedByteBuf.java:205)
at io.vertx.core.buffer.impl.BufferImpl.getByte(BufferImpl.java:74)
at io.vertx.redis.impl.ReplyParser.parseResult(ReplyParser.java:131)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:173)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:21)
at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:308)
at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$28(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler$$Lambda$56/1085121179.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.lambda$wrapTask$16(ContextImpl.java:333)
at io.vertx.core.impl.ContextImpl$$Lambda$22/618696025.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:225)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:31)
at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:124)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
at java.lang.Thread.run(Thread.java:745)

Unhandled exception
java.lang.ClassCastException: io.vertx.core.buffer.impl.BufferImpl cannot be cast to [Lio.vertx.redis.impl.Reply;
at io.vertx.redis.impl.Reply.asType(Reply.java:120)
at io.vertx.redis.impl.RedisConnection.handleReply(RedisConnection.java:412)
at io.vertx.redis.impl.RedisConnection$$Lambda$26/293542278.handle(Unknown Source)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:179)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:21)
at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:308)
at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$28(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler$$Lambda$56/1085121179.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.lambda$wrapTask$16(ContextImpl.java:333)
at io.vertx.core.impl.ContextImpl$$Lambda$22/618696025.run(Unknown Source)
at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:225)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:54)
at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:31)
at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:124)
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
at java.lang.Thread.run(Thread.java:745)
'''

java.lang.ClassCastException, While using hgetall in a transaction

Snippet code:

client = RedisClient.create(vertx, new RedisOptions().setAddress("127.0.0.1").setPort(6379));
client.multi(multiEvent -> {
  client.hgetall("hash1", hgetallEvent -> {
    client.exec(execEvent -> /* DO ANYTHING */);
  });
});

hgetall command inside a transaction always generates a CastException.

SEVERE: Unhandled exception
java.lang.ClassCastException: java.lang.String cannot be cast to [Lio.vertx.redis.impl.Reply;
    at io.vertx.redis.impl.Reply.asType(Reply.java:124)
    at io.vertx.redis.impl.Reply.asType(Reply.java:156)
    at io.vertx.redis.impl.RedisConnection.handleReply(RedisConnection.java:376)
    at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:180)
    at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:21)
    at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:313)
    at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$30(VertxNetHandler.java:54)
    at io.vertx.core.impl.ContextImpl.lambda$wrapTask$18(ContextImpl.java:333)
    at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:225)
    at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:54)
    at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:31)
    at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:124)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:318)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:304)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
    at java.lang.Thread.run(Thread.java:745)

Implement rxjava.redis.RedisClient's observables as COLD observables instead of HOT ones

Implement rxjava.redis.RedisClient's observables as COLD observables instead of HOT ones.
It will be much better for lazy execute, free to retry and easy to make sophisicated observable combination.

service descriptor json file naming

Please note service descriptor json file name needs to match maven coords for vertx-maven-service-factory dynamic deployment to work correctly, see recent vertx-dev discussion

redisClient.setBinary should take a byte[] or a Buffer not a String (reference jedis)

byte [] value ....
        redisClient.setBinary(key, value, event -> {
            if (event.failed()) {
                confirmation.onError(event.cause());
            } else {
                confirmation.accept(true);
            }
        });

redisClient.setBinaryWithOptions method is missing (ref jedis)

        final SetOptions setOptions = new SetOptions();
        setOptions.setPX(expiry.toMillis());

        redisClient.setBinaryWithOptions(key, value,  setOptions, event -> {
            if (event.failed()) {
                confirmation.onError(event.cause());
            } else {
                confirmation.accept(true);
            }
        });

add missing geo api

http://redis.io/commands#geo

ClassCastException when using RangeLimitOptions

Simply :

RangeLimitOptions options = new RangeLimitOptions();
options.setLimit(0,1);
redis.zrangebyscore("yourkey", "-inf", "+inf", options, result -> {
    // whatever
});

Will fail with a ClassCastException : Long cannot be cast to String.

Because of the cast here : https://github.com/vert-x3/vertx-redis-client/blob/master/src/main/java/io/vertx/redis/op/RangeLimitOptions.java#L49

which is invoked on two Longs : https://github.com/vert-x3/vertx-redis-client/blob/master/src/main/java/io/vertx/redis/op/LimitOptions.java#L49-L50

Maybe we could simply fix it by replacing :

  public JsonArray toJsonArray() {
    JsonArray json = new JsonArray();
    if (withscores != null && withscores) {
      json.add("WITHSCORES");
    }

    for (Object item: super.toJsonArray()) {
      json.add((String) item);
    }
    return json;
  }

with

  public JsonArray toJsonArray() {
    JsonArray json = new JsonArray();
    if (withscores != null && withscores) {
      json.add("WITHSCORES");
    }

    for (Object item: super.toJsonArray()) {
      if (item != null) {
          json.add(item.toString());
      }
    }
    return json;
  }

But idk, maybe we just don't need strings everytime ? :\

brpop/blpop handler was not called after timeout

service.brpop("k1",2, res ->{
        if (res.succeeded()){
            if (res.result()!=null){
                //vertx.eventBus().send("data",res.result());
                System.out.println(res.result());
            }
               readData();  
        }else{
            res.cause().printStackTrace();
        }
});

I am using blocking command like that, but the handler was not called after redis timeout.

Redis client handler is not being called.

Version

• vert.x core:3.3.2
• vert.x redis client:3.3.2

Context

Hi,
We are calling the following code in execute blocking in event loop
redisClient.setex(demandFactorRedisKeyPrefix + key, keyExpiry, value, r -> {
log.info("inside handler");
}

We have 8 verticles and redisclient is a static variable shared across all verticle instances. Redisclient is initialized by passing one of the vertx instances.
We are noticing that sometimes the handler is not called.

Can you please help us in debugging the issue.

add missing commands

Most of the 3.x version commands are missing (these regard to cluster management mostly)

java.lang.ClassCastException: java.lang.Long cannot be cast to [Lio.vertx.redis.impl.Reply;

From @hansen2014 on June 28, 2016 9:58

RedisTransaction rtx = redisClient.transaction();

    rtx.multi(r -> {

        String setKey = RepositoryHelper.createKSKey("SessionScreenSymbol", clientCd, sessionId, viewNum);

        List<String> keyList = new ArrayList<>();
        for (String data : delS) {
            keyList.add(data);
            rtx.del(data, r1 -> {
                GLog.info("trans del executing ..., result: {} ", r1);
            });
        }

        if (delS.size() > 0) {
            rtx.sremMany(setKey, keyList, rst -> {
                GLog.info("trans strm executing ..., result: {}", rst);
            });
        }

        rtx.exec(handler);
    });

sometimes, will throw exception like that , and then, can not receive the reply. :
fatal: Unhandled exception
java.lang.ClassCastException: java.lang.Long cannot be cast to [Lio.vertx.redis.impl.Reply;
at io.vertx.redis.impl.Reply.asType(Reply.java:94)
at io.vertx.redis.impl.Reply.asType(Reply.java:156)
at io.vertx.redis.impl.RedisConnection.handleReply(RedisConnection.java:441)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:180)
at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:21)
at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:312)
at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$0(VertxNetHandler.java:67)
at io.vertx.core.impl.ContextImpl.lambda$wrapTask$3(ContextImpl.java:357)
at io.vertx.core.impl.OrderedExecutorFactory$OrderedExecutor.lambda$new$0(OrderedExecutorFactory.java:94)
at java.util.concurrent.ThreadPoolExecutor.runWorker(Unknown Source)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
at java.lang.Thread.run(Unknown Source)

2016-06-28 17:57:57.558 [vert.x-eventloop-thread-7] ERROR EAGLE.DEFAULT - system error.
io.vertx.core.eventbus.ReplyException: Timed out after waiting 5000(ms) for a reply. address: 136
at io.vertx.core.eventbus.impl.HandlerRegistration.sendAsyncResultFailure(HandlerRegistration.java:118) ~[vertx-core-3.3.0.jar:na]
at io.vertx.core.eventbus.impl.HandlerRegistration.lambda$new$0(HandlerRegistration.java:65) ~[vertx-core-3.3.0.jar:na]
at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle(VertxImpl.java:769) ~[vertx-core-3.3.0.jar:na]
at io.vertx.core.impl.VertxImpl$InternalTimerHandler.handle(VertxImpl.java:740) ~[vertx-core-3.3.0.jar:na]
at io.vertx.core.impl.ContextImpl.lambda$wrapTask$3(ContextImpl.java:359) ~[vertx-core-3.3.0.jar:na]
at io.netty.util.concurrent.SingleThreadEventExecutor.runAllTasks(SingleThreadEventExecutor.java:339) ~[netty-common-4.1.1.Final.jar:4.1.1.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:393) ~[netty-transport-4.1.1.Final.jar:4.1.1.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:742) ~[netty-common-4.1.1.Final.jar:4.1.1.Final]
at java.lang.Thread.run(Unknown Source) ~[na:1.8.0_91]

Copied from original issue: eclipse-vertx/vert.x#1482

`RedisTransaction.smembers` has incorrect handler type

Version

• vert.x core: 3.3.3
• vert.x redis client: 3.3.3

Context

should be Handler<AsyncResult<JsonArray>> but is Handler<AsyncResult<String>>

https://github.com/vert-x3/vertx-redis-client/blob/master/src/main/java/io/vertx/redis/RedisTransaction.java#L1982

RedisClient.smembers is correct

https://github.com/vert-x3/vertx-redis-client/blob/master/src/main/java/io/vertx/redis/impl/RedisClientImpl.java#L1075

Extra

BTW, what is the point of RedisTransaction? Looks like its apis do nothing more than corresponding RedisClient apis.

Is there a plan to support SSL connection?

For production usage, we need to connect redis with SSL connection, but seems like vert.x's redis client not supporting this feature?

Fail more gracefully if start is not called

If you don't call start on the service before performing any operation, you get a strange behaviour.

For example with scriptLoad you'll get a NPE because encoding is not set (it must be set in the start method).

Here's a simple reproducer : https://github.com/aesteve/redis-client-bug

The working code is commented, if you run the main verticle it throws a NPE.

_buffer in ReplyParse is not emptied at the end of the reply

I was doing some profiling on an application and I noticed several instances of rather large byte[] arrays that aligned with the size of the object that I was retrieving from redis in ReplyParse. Here's a screenshot from the profiler.

Is it a conscious decision not to reset _buffer after a full reply has been received?

Eval/EvalSha and number of keys/args

When using eval or evalSha, the client checks that the list of keys and arguments are actually the same size.

I can't find a place in Redis documentation stating that it's a requirement. I think people should be able to specify a list of N keys and M values when using eval or evalSha.

With Vertx's redisClient, this fails because of :

https://github.com/vert-x3/vertx-redis-client/blob/master/src/main/java/io/vertx/redis/impl/RedisClientImpl.java#L240

Example using jedis :

Jedis jedis = jedisPool.getResource();
List<String> keys = new ArrayList<String>(2);
List<String> values = new ArrayList<String>(1);
keys.add("key1");
keys.add("key2");
values.add("value1");
jedis.evalSha("myScript", keys, values); // works

setex takes ttl as a long, expire as an int

Version

• vert.x core: 3.3.3
• vert.x redis client: 3.3.3

Context

The setex method requires that the number of seconds until expiry is provided as a long however the expire method requires an int; maybe there's a good reason for this that I'm missing but it seems inconsistent (in my case I have a object with a getTtl method which returns a long (I used setex first) but I'm having to cast the result to pass into expire).

[Question] Redis Pub/Sub

Does vertx-redis-client support correctly the pub/sub system of redis.
I have some issues with the subscription part of the system.

     vertx.eventBus().<JsonObject>consumer("redis.sub.channel1", received -> {
        // do whatever you need to do with your message
        JsonObject value = received.body().getJsonObject("value");
        // the value is a JSON doc with the following properties
        // channel - The channel to which this message was sent
        // pattern - Pattern is present if you use psubscribe command and is the pattern that matched this message channel
        // message - The message payload
        System.out.println(value.encodePrettily());
    });

    RedisClient redis = RedisClient.create(vertx, new JsonObject());

    redis.subscribe(Collections.singletonList("channel1"), res -> {
        if (res.succeeded()) {
            // so something...
            RedisClient redis2 = RedisClient.create(vertx, new JsonObject());

            redis2.publish("channel1", "Hello World!", res2 -> {
                if (res2.succeeded()) {
                    // so something...
                }
            });
        }
    });

Publish system seems to work but the message is lost ?

Redisclient throwing ArrayIndexOutOfBoundsExceptions under heavy load in 3.1

Hi,

Yesterday I upgraded my project to use 3.1 but currently running into problems with code that was working fine in 3.0.

At the moment, when under heavy load or sometimes at random, heavy redis operations are throwing exceptions. Like zcan,zrange,lrange etc.

I haven't created a reproducer but the code I am testing with is so basic and stripped down I can just post it here:

private static int counter = 0;

    public static void main(String[] args) {

        Vertx vertx = Vertx.vertx();

        RedisOptions config = new RedisOptions();
        config.setHost("192.168.99.100");
        config.setPort(6379);
        RedisClient redisClient =  new io.vertx.rxjava.redis.RedisClient(io.vertx.redis.RedisClient.create(vertx, config));

        int numberOfIterations = 200000;

        System.out.println("started");
        for (int i = 0; i < numberOfIterations; i++) {
            redisClient.zscan("gm:1:lb:0:rm:b:players","0", ScanOptions.NONE,event ->
            {
                if(event.succeeded())
                {
                }
            });
        }

    }

Error I am getting is

10 14, 2015 11:26:11 午前 io.vertx.core.impl.ContextImpl
重大: Unhandled exception
java.lang.ArrayIndexOutOfBoundsException: 13141
    at io.netty.buffer.UnpooledHeapByteBuf._getByte(UnpooledHeapByteBuf.java:292)
    at io.netty.buffer.UnpooledHeapByteBuf.getByte(UnpooledHeapByteBuf.java:287)
    at io.netty.buffer.WrappedByteBuf.getByte(WrappedByteBuf.java:205)
    at io.vertx.core.buffer.impl.BufferImpl.getByte(BufferImpl.java:74)
    at io.vertx.redis.impl.ReplyParser.parseResult(ReplyParser.java:131)
    at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:173)
    at io.vertx.redis.impl.ReplyParser.handle(ReplyParser.java:21)
    at io.vertx.core.net.impl.NetSocketImpl.handleDataReceived(NetSocketImpl.java:308)
    at io.vertx.core.net.impl.VertxNetHandler.lambda$channelRead$28(VertxNetHandler.java:54)
    at io.vertx.core.net.impl.VertxNetHandler$$Lambda$20/699439032.run(Unknown Source)
    at io.vertx.core.impl.ContextImpl.lambda$wrapTask$16(ContextImpl.java:333)
    at io.vertx.core.impl.ContextImpl$$Lambda$14/1134865005.run(Unknown Source)
    at io.vertx.core.impl.ContextImpl.executeFromIO(ContextImpl.java:225)
    at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:54)
    at io.vertx.core.net.impl.VertxNetHandler.channelRead(VertxNetHandler.java:31)
    at io.vertx.core.net.impl.VertxHandler.channelRead(VertxHandler.java:124)
    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:308)
    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:294)
    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:846)
    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:131)
    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:511)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:468)
    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:382)
    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:354)
    at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:112)
    at java.lang.Thread.run(Thread.java:745)

I am also getting a lot of message handler timeout errors but I figure it is maybe related to this problem.

Kind regards,

Yves

brpoplpush should have Handler<AsyncResult<JsonObject>> parameter

The current implementation for brpoplpush has Handler<AsyncResult> as it's last parameter. With existing singniture, there is no way to get the popped element in a single atomic operation.

The api should have the Handler<AsyncResult> parameter instead.

Make usable from multiple contexts

The current service is not thread safe and makes the assumption that it is always accessed from same context.

We should make it usable from different contexts.

support timeout option

Does vertx-redis-client support timeout? I need redis command to be finish in some milliseconds, how can I do it?

Eval not returning object type of result

In Jedis client, Eval op will return Object. But in redis-client here we return nothing with a void type. Any specific reason that we could not return object type? I could work on a PR if we want the eval and eval sha to be able to return object type data.

Thanks.