diff --git a/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/LockRequestHandlerAdvice.java b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/LockRequestHandlerAdvice.java new file mode 100644 index 0000000000..52b965ae24 --- /dev/null +++ b/spring-integration-core/src/main/java/org/springframework/integration/handler/advice/LockRequestHandlerAdvice.java @@ -0,0 +1,206 @@ +/* + * Copyright 2025 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.integration.handler.advice; + +import java.time.Duration; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.locks.Lock; +import java.util.function.Function; + +import org.springframework.expression.EvaluationContext; +import org.springframework.expression.Expression; +import org.springframework.integration.expression.ExpressionUtils; +import org.springframework.integration.expression.FunctionExpression; +import org.springframework.integration.expression.ValueExpression; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.integration.support.utils.IntegrationUtils; +import org.springframework.lang.Nullable; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.util.Assert; + +/** + * The {@link AbstractRequestHandlerAdvice} to ensure exclusive access to the + * {@code AbstractReplyProducingMessageHandler.RequestHandler#handleRequestMessage(Message)} calls + * based on the {@code lockKey} from message. + *

+ * If {@code lockKey} for the message is {@code null}, the no locking around the call. + * However, if {@link } + * + * @author Artem Bilan + * + * @since 6.5 + */ +public class LockRequestHandlerAdvice extends AbstractRequestHandlerAdvice { + + private final LockRegistry lockRegistry; + + private final Expression lockKeyExpression; + + @Nullable + private MessageChannel discardChannel; + + @Nullable + private Expression waitLockDurationExpression; + + private EvaluationContext evaluationContext; + + /** + * Construct an advice instance based on a {@link LockRegistry} and fixed (shared) lock key. + * @param lockRegistry the {@link LockRegistry} to use. + * @param lockKey the static (shared) lock key for all the calls. + */ + public LockRequestHandlerAdvice(LockRegistry lockRegistry, Object lockKey) { + this(lockRegistry, new ValueExpression<>(lockKey)); + } + + /** + * Construct an advice instance based on a {@link LockRegistry} + * and SpEL expression for the lock key against request message. + * @param lockRegistry the {@link LockRegistry} to use. + * @param lockKeyExpression the SpEL expression to evaluate a lock key against request message. + */ + public LockRequestHandlerAdvice(LockRegistry lockRegistry, Expression lockKeyExpression) { + Assert.notNull(lockRegistry, "'lockRegistry' must not be null"); + Assert.notNull(lockKeyExpression, "'lockKeyExpression' must not be null"); + this.lockRegistry = lockRegistry; + this.lockKeyExpression = lockKeyExpression; + } + + /** + * Construct an advice instance based on a {@link LockRegistry} + * and function for the lock key against request message. + * @param lockRegistry the {@link LockRegistry} to use. + * @param lockKeyFunction the function to evaluate a lock key against request message. + */ + public LockRequestHandlerAdvice(LockRegistry lockRegistry, Function, Object> lockKeyFunction) { + Assert.notNull(lockRegistry, "'lockRegistry' must not be null"); + Assert.notNull(lockKeyFunction, "'lockKeyFunction' must not be null"); + this.lockRegistry = lockRegistry; + this.lockKeyExpression = new FunctionExpression<>(lockKeyFunction); + } + + /** + * Optional duration for a {@link Lock#tryLock(long, TimeUnit)} API. + * Otherwise, {@link Lock#lockInterruptibly()} is used. + * @param waitLockDuration the duration for {@link Lock#tryLock(long, TimeUnit)}. + */ + public void setWaitLockDuration(Duration waitLockDuration) { + setWaitLockDurationExpression(new ValueExpression<>(waitLockDuration)); + } + + /** + * The SpEL expression to evaluate a {@link Lock#tryLock(long, TimeUnit)} duration + * against request message. + * Can be evaluated to {@link Duration}, {@code long} (with meaning as milliseconds), + * or to string in the duration ISO-8601 format. + * @param waitLockDurationExpression SpEL expression for duration. + */ + public void setWaitLockDurationExpression(Expression waitLockDurationExpression) { + this.waitLockDurationExpression = waitLockDurationExpression; + } + + /** + * The SpEL expression to evaluate a {@link Lock#tryLock(long, TimeUnit)} duration + * against request message. + * Can be evaluated to {@link Duration}, {@code long} (with meaning as milliseconds), + * or to string in the duration ISO-8601 format. + * @param waitLockDurationExpression SpEL expression for duration. + */ + public void setWaitLockDurationExpressionString(String waitLockDurationExpression) { + this.waitLockDurationExpression = EXPRESSION_PARSER.parseExpression(waitLockDurationExpression); + } + + /** + * The function to evaluate a {@link Lock#tryLock(long, TimeUnit)} duration + * against request message. + * @param waitLockDurationFunction the function for duration. + */ + public void setWaitLockDurationFunction(Function, Duration> waitLockDurationFunction) { + this.waitLockDurationExpression = new FunctionExpression<>(waitLockDurationFunction); + } + + /** + * Set a channel where to send a message for which {@code lockKey} is evaluated to {@code null}. + * If this is not set and {@code lockKey == null}, no locking around the call. + * @param discardChannel the channel to send messages without a key. + */ + public void setDiscardChannel(@Nullable MessageChannel discardChannel) { + this.discardChannel = discardChannel; + } + + @Override + protected void onInit() { + super.onInit(); + this.evaluationContext = ExpressionUtils.createStandardEvaluationContext(getBeanFactory()); + } + + @Override + protected Object doInvoke(ExecutionCallback callback, Object target, Message message) { + Object lockKey = this.lockKeyExpression.getValue(this.evaluationContext, message); + if (lockKey != null) { + Duration waitLockDuration = getWaitLockDuration(message); + try { + if (waitLockDuration == null) { + return lockRegistry.executeLocked(lockKey, callback::execute); + } + else { + return lockRegistry.executeLocked(lockKey, waitLockDuration, callback::execute); + } + } + catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "The lock for message was interrupted", ex); + } + catch (TimeoutException ex) { + throw IntegrationUtils.wrapInHandlingExceptionIfNecessary(message, + () -> "Could not acquire the lock in time: " + waitLockDuration, ex); + } + } + else { + if (this.discardChannel != null) { + this.discardChannel.send(message); + return null; + } + else { + return callback.execute(); + } + } + } + + @Nullable + private Duration getWaitLockDuration(Message message) { + if (this.waitLockDurationExpression != null) { + Object value = this.waitLockDurationExpression.getValue(this.evaluationContext, message); + if (value != null) { + if (value instanceof Duration duration) { + return duration; + } + else if (value instanceof Long aLong) { + return Duration.ofMillis(aLong); + } + else { + return Duration.parse(value.toString()); + } + } + } + return null; + } + +} diff --git a/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/LockRequestHandlerAdviceTests.java b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/LockRequestHandlerAdviceTests.java new file mode 100644 index 0000000000..7350631581 --- /dev/null +++ b/spring-integration-core/src/test/java/org/springframework/integration/handler/advice/LockRequestHandlerAdviceTests.java @@ -0,0 +1,139 @@ +/* + * Copyright 2025 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.integration.handler.advice; + +import java.util.concurrent.ExecutionException; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.integration.annotation.ServiceActivator; +import org.springframework.integration.channel.QueueChannel; +import org.springframework.integration.config.EnableIntegration; +import org.springframework.integration.core.AsyncMessagingTemplate; +import org.springframework.integration.support.MessageBuilder; +import org.springframework.integration.support.locks.DefaultLockRegistry; +import org.springframework.integration.support.locks.LockRegistry; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.core.MessagePostProcessor; +import org.springframework.messaging.support.GenericMessage; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * @author Artem Bilan + * + * @since 6.5 + */ +@SpringJUnitConfig +@DirtiesContext +public class LockRequestHandlerAdviceTests { + + private static final String LOCK_KEY_HEADER = "lock-key-header"; + + @Autowired + MessageChannel inputChannel; + + @Autowired + QueueChannel discardChannel; + + @Test + void verifyLockAroundHandler() throws ExecutionException, InterruptedException, TimeoutException { + AsyncMessagingTemplate messagingTemplate = new AsyncMessagingTemplate(); + MessagePostProcessor messagePostProcessor = + message -> + MessageBuilder.fromMessage(message) + .setHeader(LOCK_KEY_HEADER, "someLock") + .build(); + Future test1 = + messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "test1", messagePostProcessor); + Future test2 = + messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "test2", messagePostProcessor); + + assertThat(test1.get(10, TimeUnit.SECONDS)).isEqualTo("test1-1"); + assertThat(test2.get(10, TimeUnit.SECONDS)).isEqualTo("test2-1"); + + messagingTemplate.send(this.inputChannel, new GenericMessage<>("no_lock_key")); + + Message receive = this.discardChannel.receive(10_000); + assertThat(receive) + .extracting(Message::getPayload) + .isEqualTo("no_lock_key"); + + Future test3 = + messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "longer_process", messagePostProcessor); + + Future test4 = + messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "test4", messagePostProcessor); + + assertThat(test3.get(10, TimeUnit.SECONDS)).isEqualTo("longer_process-1"); + + assertThat(test4).failsWithin(10, TimeUnit.SECONDS) + .withThrowableOfType(ExecutionException.class) + .withRootCauseInstanceOf(TimeoutException.class) + .withStackTraceContaining("Could not acquire the lock in time: PT1S"); + } + + @Configuration + @EnableIntegration + public static class Config { + + @Bean + LockRegistry lockRegistry() { + return new DefaultLockRegistry(); + } + + @Bean + QueueChannel discardChannel() { + return new QueueChannel(); + } + + @Bean + LockRequestHandlerAdvice lockRequestHandlerAdvice(LockRegistry lockRegistry, QueueChannel discardChannel) { + LockRequestHandlerAdvice lockRequestHandlerAdvice = + new LockRequestHandlerAdvice(lockRegistry, (message) -> message.getHeaders().get(LOCK_KEY_HEADER)); + lockRequestHandlerAdvice.setDiscardChannel(discardChannel); + lockRequestHandlerAdvice.setWaitLockDurationExpressionString("'PT1s'"); + return lockRequestHandlerAdvice; + } + + AtomicInteger counter = new AtomicInteger(); + + @ServiceActivator(inputChannel = "inputChannel", adviceChain = "lockRequestHandlerAdvice") + String handleWithDelay(String payload) throws InterruptedException { + int currentCount = this.counter.incrementAndGet(); + Thread.sleep("longer_process".equals(payload) ? 2000 : 500); + try { + return payload + "-" + currentCount; + } + finally { + this.counter.decrementAndGet(); + } + } + + } + +} diff --git a/src/reference/antora/modules/ROOT/pages/handler-advice/classes.adoc b/src/reference/antora/modules/ROOT/pages/handler-advice/classes.adoc index 0f199defaf..4c8df69cf6 100644 --- a/src/reference/antora/modules/ROOT/pages/handler-advice/classes.adoc +++ b/src/reference/antora/modules/ROOT/pages/handler-advice/classes.adoc @@ -10,6 +10,7 @@ In addition to providing the general mechanism to apply AOP advice classes, Spri * `CacheRequestHandlerAdvice` (described in xref:handler-advice/classes.adoc#cache-advice[Caching Advice]) * `ReactiveRequestHandlerAdvice` (described in xref:handler-advice/reactive.adoc[Reactive Advice]) * `ContextHolderRequestHandlerAdvice` (described in xref:handler-advice/context-holder.adoc[Context Holder Advice]) +* `LockRequestHandlerAdvice` (described in xref:handler-advice/lock.adoc[Lock Advice]) [[expression-advice]] diff --git a/src/reference/antora/modules/ROOT/pages/handler-advice/lock.adoc b/src/reference/antora/modules/ROOT/pages/handler-advice/lock.adoc new file mode 100644 index 0000000000..a6ecde830a --- /dev/null +++ b/src/reference/antora/modules/ROOT/pages/handler-advice/lock.adoc @@ -0,0 +1,49 @@ +[[lock-advice]] += Lock Advice + +Starting with version 6.5, the `LockRequestHandlerAdvice` has been introduced. +This advice evaluates a lock key against request message and performs `LockRegistry.executeLocked()` API. +The goal of the advice is to achieve exclusive access to the service invocation according to the `lockKey` context, meaning that different keys may still get concurrent access to the service. + +The `LockRequestHandlerAdvice` requires a xref:distributed-locks.adoc[LockRegistry], and a static, SpEL or function-based lock key callback. +If `lockKey` is evaluated to `null`, no locking is held around service call. +However, a `discardChannel` can be provided - and such a message with null key will be sent to this channel instead. +Also, a `waitLockDuration` option can be provided to use `Lock.tryLock(long, TimeUnit)` API instead of `Lock.lockInterruptibly()`. + +Following is a sample how a `LockRequestHandlerAdvice` can be used: + +[source, java] +---- +@Bean +LockRegistry lockRegistry() { + return new DefaultLockRegistry(); +} + +@Bean +QueueChannel discardChannel() { + return new QueueChannel(); +} + +@Bean +LockRequestHandlerAdvice lockRequestHandlerAdvice(LockRegistry lockRegistry, QueueChannel discardChannel) { + LockRequestHandlerAdvice lockRequestHandlerAdvice = + new LockRequestHandlerAdvice(lockRegistry, (message) -> message.getHeaders().get(LOCK_KEY_HEADER)); + lockRequestHandlerAdvice.setDiscardChannel(discardChannel); + lockRequestHandlerAdvice.setWaitLockDurationExpressionString("'PT1s'"); + return lockRequestHandlerAdvice; +} + +AtomicInteger counter = new AtomicInteger(); + +@ServiceActivator(inputChannel = "inputChannel", adviceChain = "lockRequestHandlerAdvice") +String handleWithDelay(String payload) throws InterruptedException { + int currentCount = this.counter.incrementAndGet(); + Thread.sleep("longer_process".equals(payload) ? 2000 : 500); + try { + return payload + "-" + currentCount; + } + finally { + this.counter.decrementAndGet(); + } +} +---- \ No newline at end of file diff --git a/src/reference/antora/modules/ROOT/pages/whats-new.adoc b/src/reference/antora/modules/ROOT/pages/whats-new.adoc index 12786c14a2..3ca598cf34 100644 --- a/src/reference/antora/modules/ROOT/pages/whats-new.adoc +++ b/src/reference/antora/modules/ROOT/pages/whats-new.adoc @@ -31,6 +31,12 @@ The `AbstractMessageChannel` beans now throw a special `MessageDispatchingExcept In general, it is a design error to try to produce a message from `afterPropertiesSet()`, `@PostConstruct` or bean definition methods. The `SmartLifecycle.start()` is preferred way for this kind of logic, or better to do that via inbound channel adapters. +[[x6.5-lock-request-handler-advice]] +== The `LockRequestHandlerAdvice` + +A new `LockRequestHandlerAdvice` is introduced to keep the lock for a key based on a request message for message handler invocation. +See xref:handler-advice.adoc[] for more information. + [[x6.5-correlation-changes]] == The `discardIndividuallyOnExpiry` Option For Correlation Handlers