spring-projects / spring-projects/spring-data-redis
Support for re-using the same connection when we set "block" greater than 0 in XREADGROUP(using lettuce lib underneath)
@mp911de is already working on this.
Since Feb 7, 2022.
- Dominant language
- Java
- Stars
- 1.9k
- Forks
- 1.3k
- PR merge metrics
- No merged PRs in 30d
Description
We are continuously listening for messages on a stream. When we use block > 0(We do not want to set it to 0) in the XREADGROUP command, spring is creating new connections for every read. This results in so many TIME_WAIT ports on the client which builds up rapidly if there are many incoming messages. For our use case, we need to set block > 0 and also re-use the same connection. Checking the code in DefaultStreamReceiver's constructor, looks like overriding the StreamReadOptions.isBlocking() to always return false will solve our problem but the DefaultStreamReceiver does not take a custom StreamReadOptions. Can you expose a method like setReadOptions(reaOptions) on DefaultStreamReceiver to take it as an argument?
Sample code I use:
StreamReceiverOptions<String, MapRecord<String, String, String>> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofSeconds(30)).build();
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(factory, options);
receiver.receiveAutoAck(Consumer.from("group", "consumer"), StreamOffset.create("stream", ReadOffset.lastConsumed()))//
.flatMap(record -> Mono.fromCallable(() -> consume(record)).subscribeOn(Schedulers.boundedElastic()))//
.subscribe();
Contributor guide
First steps
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- Open a pull request that references the issue number.
Assessment
This issue has not been assessed yet.