Enable `ReactiveRedisTemplate` for multi-tenancy usage

[petromir]

Hi,

I would like to use ReactiveRedisTemplate in multitenant environment. In short, the option to use single ReactiveRedisTemplate which can connect to multiple Redis instances based on tenant values passed in reactive context.
Unfortunately, I can not connect the reactive context with the connection factory so I can create different connections (or using different connection pool) based on tenant value in the context.

One option is to override doInConnection, but it has private modifier. Changing it will allow me to provide custom logic about connection creation.

[mp911de]

doInConnection is an internal method that handles resource allocation. What we should look for is how to enable ReactiveRedisTemplate for multi-tenancy usage. The first part of resource allocation is having a Mono emitting ReactiveRedisConnection. We could introduce a Mono<ReactiveRedisConnection> getConnection() method as extension point.

[petromir]

Thanks a lot. When do you plan to tackle this?

[mp911de]

Having a pull request would improve things in terms of timing so we could integrate it faster.

[petromir]

I knew you gonna say that 😄

[petromir]

We could introduce a Mono getConnection() method as extension point.

I assume you mean in having this in ReactiveRedisTemplate?

[mp911de]

Yes, exactly. A protected method that can be overridden by subclasses. Right now, the calls to obtain a connection return plain ReactiveConnection on the connection factory and are not suitable to access the reactor context. So solving the issue within the template would create the least impact.

[petromir]

Something like

private <T> Publisher<T> doInConnection(ReactiveRedisCallback<T> action, boolean exposeConnection) {

	Assert.notNull(action, "Callback object must not be null");

	return Flux.usingWhen(Mono.deferContextual(contextView -> getConnection(contextView, exposeConnection)), conn -> {
		Publisher<T> result = action.doInRedis(conn);

		return postProcessResult(result, conn, false);

 	}, ReactiveRedisConnection::closeLater);
}

protected Mono<ReactiveRedisConnection> getConnection(final ContextView contextView, boolean exposeConnection) {
	ReactiveRedisConnectionFactory factory = getConnectionFactory();
	ReactiveRedisConnection conn = factory.getReactiveConnection();
	ReactiveRedisConnection connToUse = preProcessConnection(conn, false);

	return Mono.just(exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
}

[mp911de]

Mono.fromSupplier() is sufficient. One calling getConnection() that returns a Mono would expect deferred resource allocation. Otherwise, you already have a connection allocated and the only way to dispose it is by subscribing to the mono. Retaining fromSupplier(…) avoids such leaks in the first place.

[petromir]

Then how I will be able to pass the context to getConnection()?

[mp911de]

Since the method returns a Mono already, you can override the method by calling Mono.deferContextual(…) in your override.

[petromir]

You mean

private <T> Publisher<T> doInConnection(ReactiveRedisCallback<T> action, boolean exposeConnection) {

	Assert.notNull(action, "Callback object must not be null");

	return Flux.usingWhen(getConnection(exposeConnection), conn -> {
		Publisher<T> result = action.doInRedis(conn);

		return postProcessResult(result, conn, false);

 	}, ReactiveRedisConnection::closeLater);
}

protected Mono<ReactiveRedisConnection> getConnection(boolean exposeConnection) {
	ReactiveRedisConnectionFactory factory = getConnectionFactory();
	ReactiveRedisConnection conn = factory.getReactiveConnection();
	ReactiveRedisConnection connToUse = preProcessConnection(conn, false);

	return Mono.just(exposeConnection ? connToUse : createRedisConnectionProxy(connToUse));
}

Then in my overriding class

protected Mono<ReactiveRedisConnection> getConnection(boolean exposeConnection) {
	return Mono.deferContextual(contextView -> {
	    // Something that returns Mono<ReactiveRedisConnection> based on the value in the context
        }
)