reactor.rabbitmq.Utils.cache could cause memory leak

[robotmrv]

reactor.rabbitmq.Utils.cache() and methods which use it could cause memory leak.
For details please see reactor/reactor-core#1916

Expected Behavior

Utils#cache() should not submit invalidation task for "infinite" time on success, but just cache result.

Actual Behavior

Utils#cache() submits invalidation task with delay Duration.ofMillis(Long.MAX_VALUE) i.e. "infinite" time on success, which could cause memory leak

Steps to Reproduce

make a lot of cache(myMono).subscribe() and invalidation tasks will never be garbage collected.

Possible Solution

see related reactor-core issue reactor/reactor-core#1916
or if you do not see issue here just warn in javadocs that Utils#cache() or Sender.rpcClient() could cause memory leaks if they are invoked many times.

Your Environment

• Reactor version(s) used: reactor-rabbitmq:1.3.0.RELEASE
• Other relevant libraries versions (eg. netty, ...):
• JVM version (javar -version):
• OS and version (eg uname -a):

[simonbasle]

Again, I don't think that "make a lot of cache(MyMono).subscribe() is a relevant pattern, and this can probably only be fixed in core, so this doesn't feel too bad. I'm opening a PR on the core issue though, as a general low-criticality enhancement.

[acogoluegnes]

This PR in Reactor Core fixes the memory leak problem for Duration.ofMillis(Long.MAX_VALUE), that is for Utils#cache.

[robotmrv]

@simonbasle
I am happy with this solution

I'm opening a PR on the core issue though, as a general low-criticality enhancement.

but unfortunately

I don't think that "make a lot of cache(MyMono).subscribe() is a relevant pattern

this is my pattern, because we have dynamic routing key in RPC so execution is something like

• Web Endpoint (it could be many requests) ->
• do some stuff ->
• create new RabbitMQ RPC client and call it ->
• response Ok

So Utils.cache() caches channel and submits to Schedulers.parallel() invalidation task which will never be evicted, and will hold channel and all other stuff from MonoCacheTime till application shutdown

[acogoluegnes]

@robotmrv The PR mentioned above will not schedule any invalidation task as soon as Duration.ofMillis(Long.MAX_VALUE) is used, which is the case in Utils#cache. So no task should leak. Does that address your concern?

[robotmrv]

@acogoluegnes I've just pointed out that there is a case where current behavior could lead to significant leak and it should be fixed.
I am ok and happy with solution from the PR