diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java index ad0f204d5..077746ed3 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisOperations.java @@ -63,6 +63,23 @@ public interface ReactiveRedisOperations { */ Flux execute(ReactiveRedisCallback action); + /** + * Executes the given action within a Redis session using the same + * {@link org.springframework.data.redis.connection.ReactiveRedisConnection}. Application exceptions thrown by the + * action object get propagated to the caller (can only be unchecked) whenever possible. Redis exceptions are + * transformed into appropriate DAO ones. Allows for returning a result object, that is a domain object or a + * collection of domain objects. Performs automatic serialization/deserialization for the given objects to and from + * binary data suitable for the Redis storage. Note: Callback code is not supposed to handle transactions itself! Use + * an appropriate transaction manager. Generally, callback code must not touch any Connection lifecycle methods, like + * close, to let the template do its work. + * + * @param return type + * @param action callback object that specifies the Redis action + * @return a result object returned by the action or {@link Flux#empty()}. + * @since 2.6 + */ + Flux executeInSession(ReactiveRedisSessionCallback action); + // ------------------------------------------------------------------------- // Methods dealing with Redis Pub/Sub // ------------------------------------------------------------------------- diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisSessionCallback.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisSessionCallback.java new file mode 100644 index 000000000..4b87ab73a --- /dev/null +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisSessionCallback.java @@ -0,0 +1,48 @@ +/* + * Copyright 2017-2021 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.data.redis.core; + +import org.reactivestreams.Publisher; +import org.springframework.dao.DataAccessException; +import org.springframework.data.redis.connection.ReactiveRedisConnection; + +/** + * Generic callback interface for code that wants to use the same {@link ReactiveRedisConnection} avoiding connection + * allocation overhead upon each Template API method call. Allows to execute any number of operations on a single + * {@link ReactiveRedisConnection}, using any type and number of commands. + *

+ * This is particularly useful for issuing multiple calls on the same connection. + * + * @param + * @author Mark Paluch + * @since 2.6 + * @see ReactiveRedisOperations#executeInSession(ReactiveRedisSessionCallback) + */ +public interface ReactiveRedisSessionCallback { + + /** + * Gets called by {@link ReactiveRedisOperations#executeInSession(ReactiveRedisSessionCallback)} with an active Redis + * connection. Does not need to care about activating or closing the {@link ReactiveRedisConnection}. + *

+ * Allows for returning a result object created within the callback, i.e. a domain object or a collection of domain + * objects. + * + * @param operations template associated with a connection. + * @return a result object publisher + * @throws DataAccessException in case of custom exceptions + */ + Publisher doWithOperations(ReactiveRedisOperations operations) throws DataAccessException; +} diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java index fba47e095..6ae24d0fc 100644 --- a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java +++ b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java @@ -151,6 +151,14 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations Flux executeInSession(ReactiveRedisSessionCallback action) { + + Assert.notNull(action, "Callback object must not be null"); + return Flux + .from(doInConnection(connection -> action.doWithOperations(withConnection(connection)), exposeConnection)); + } + /** * Create a reusable Flux for a {@link ReactiveRedisCallback}. Callback is executed within a connection context. The * connection is released outside the callback. @@ -188,7 +196,7 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations Publisher doInConnection(ReactiveRedisCallback action, boolean exposeConnection) { + Publisher doInConnection(ReactiveRedisCallback action, boolean exposeConnection) { Assert.notNull(action, "Callback object must not be null"); @@ -742,6 +750,31 @@ public class ReactiveRedisTemplate implements ReactiveRedisOperations withConnection(ReactiveRedisConnection connection) { + return new BoundConnectionRedisTemplate(connection, connectionFactory, serializationContext); + } + + class BoundConnectionRedisTemplate extends ReactiveRedisTemplate { + + private final ReactiveRedisConnection connection; + + public BoundConnectionRedisTemplate(ReactiveRedisConnection connection, + ReactiveRedisConnectionFactory connectionFactory, RedisSerializationContext serializationContext) { + super(connectionFactory, serializationContext, true); + this.connection = connection; + } + + @Override + Publisher doInConnection(ReactiveRedisCallback action, boolean exposeConnection) { + + Assert.notNull(action, "Callback object must not be null"); + + ReactiveRedisConnection connToUse = ReactiveRedisTemplate.this.preProcessConnection(connection, true); + Publisher result = action.doInRedis(connToUse); + return ReactiveRedisTemplate.this.postProcessResult(result, connToUse, true); + } + } + private ByteBuffer rawKey(K key) { return getSerializationContext().getKeySerializationPair().getWriter().write(key); } diff --git a/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt b/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt index f3023d300..89af9db12 100644 --- a/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt +++ b/src/main/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensions.kt @@ -22,7 +22,7 @@ import kotlinx.coroutines.reactive.awaitFirstOrNull import kotlinx.coroutines.reactive.awaitSingle import org.springframework.data.redis.connection.DataType import org.springframework.data.redis.connection.ReactiveRedisConnection -import org.springframework.data.redis.connection.ReactiveSubscription.* +import org.springframework.data.redis.connection.ReactiveSubscription.Message import org.springframework.data.redis.core.script.RedisScript import org.springframework.data.redis.listener.Topic import org.springframework.data.redis.serializer.RedisElementReader @@ -36,8 +36,20 @@ import java.time.Instant * @author Sebastien Deleuze * @since 2.2 */ -fun ReactiveRedisOperations.executeAsFlow(action: (ReactiveRedisConnection) -> Flow): Flow = - execute { action(it).asPublisher() }.asFlow() +fun ReactiveRedisOperations.executeAsFlow(action: (ReactiveRedisConnection) -> Flow): Flow { + return execute { action(it).asPublisher() }.asFlow() +} + +/** + * Coroutines variant of [ReactiveRedisOperations.execute]. + * + * @author Mark Paluch + * @since 2.6 + */ +fun ReactiveRedisOperations.executeInSessionAsFlow( + action: (ReactiveRedisOperations) -> Flow +): Flow = + executeInSession { action(it).asPublisher() }.asFlow() /** * Coroutines variant of [ReactiveRedisOperations.execute]. @@ -45,8 +57,12 @@ fun ReactiveRedisOperations.executeAsFlow(acti * @author Sebastien Deleuze * @since 2.2 */ -fun ReactiveRedisOperations.executeAsFlow(script: RedisScript, keys: List = emptyList(), args: List<*> = emptyList()): Flow = - execute(script, keys, args).asFlow() +fun ReactiveRedisOperations.executeAsFlow( + script: RedisScript, + keys: List = emptyList(), + args: List<*> = emptyList() +): Flow = + execute(script, keys, args).asFlow() /** * Coroutines variant of [ReactiveRedisOperations.execute]. diff --git a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt b/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt index 7e98bd1f6..26cab9292 100644 --- a/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt +++ b/src/test/kotlin/org/springframework/data/redis/core/ReactiveRedisOperationsExtensionsUnitTests.kt @@ -50,7 +50,8 @@ class ReactiveRedisOperationsExtensionsUnitTests { every { operations.execute(any>()) } returns Flux.just("foo") runBlocking { - assertThat(operations.executeAsFlow { flow { emit("foo")} }.toList()).contains("foo") + assertThat(operations.executeAsFlow { flow { emit("foo") } } + .toList()).contains("foo") } verify { @@ -58,12 +59,36 @@ class ReactiveRedisOperationsExtensionsUnitTests { } } + @Test // GH-2110 + fun `executeInSession with calllback`() { + + val operations = mockk>() + every { operations.executeInSession(any>()) } returns Flux.just( + "foo" + ) + + runBlocking { + assertThat(operations.executeInSessionAsFlow { flow { emit("foo") } } + .toList()).contains("foo") + } + + verify { + operations.executeInSession(any>()) + } + } + @Test // DATAREDIS-1033 fun `execute with script`() { val script = RedisScript.of("foo") val operations = mockk>() - every { operations.execute(any>(), any(), any()) } returns Flux.just("foo") + every { + operations.execute( + any>(), + any(), + any() + ) + } returns Flux.just("foo") runBlocking { assertThat(operations.executeAsFlow(script).toList()).contains("foo")