GH-8056: Add LockRequestHandlerAdvice

Fixes: https://github.com/spring-projects/spring-integration/issues/8056

* Add `LockRequestHandlerAdvice` for exclusive service access against specific `key`.

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.
+ * <p>
+ * 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<Message<?>, 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<Message<?>, 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;
+	}
+
+}

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<Object> test1 =
+				messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "test1", messagePostProcessor);
+		Future<Object> 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<Object> test3 =
+				messagingTemplate.asyncConvertSendAndReceive(this.inputChannel, "longer_process", messagePostProcessor);
+
+		Future<Object> 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();
+			}
+		}
+
+	}
+
+}

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]]

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();
+    }
+}
+----

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