diff --git a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java index dcb2c8f01e..6842fe089a 100644 --- a/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java +++ b/core/spring-cloud-stream/src/main/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSource.java @@ -108,35 +108,70 @@ public DefaultPollableMessageSource( public void setSource(MessageSource source) { ProxyFactory pf = new ProxyFactory(source); - class ReceiveAdvice implements MethodInterceptor { - private final List interceptors = new ArrayList<>(); + class ReceiveAdvice implements MethodInterceptor { @Override public Object invoke(MethodInvocation invocation) throws Throwable { - Object result = invocation.proceed(); - if (result instanceof Message received) { - for (ChannelInterceptor interceptor : this.interceptors) { - received = interceptor.preSend(received, DUMMY_CHANNEL); - if (received == null) { + Message result = null; + Exception completionException = null; + boolean preReceiveCompleted = false; + try { + for (ChannelInterceptor interceptor : interceptors()) { + if (!interceptor.preReceive(DUMMY_CHANNEL)) { return null; } } - return received; + preReceiveCompleted = true; + Object received = invocation.proceed(); + if (received instanceof Message message) { + result = message; + for (ChannelInterceptor interceptor : interceptors()) { + result = interceptor.postReceive(result, DUMMY_CHANNEL); + if (result == null) { + return null; + } + } + for (ChannelInterceptor interceptor : interceptors()) { + result = interceptor.preSend(result, DUMMY_CHANNEL); + if (result == null) { + return null; + } + } + } + else { + result = null; + } + return result; + } + catch (Throwable ex) { + completionException = ex instanceof Exception exception ? exception + : new IllegalStateException(ex); + throw ex; + } + finally { + if (preReceiveCompleted) { + for (ChannelInterceptor interceptor : interceptors()) { + interceptor.afterReceiveCompletion(result, DUMMY_CHANNEL, + completionException); + } + } } - return result; } } - final ReceiveAdvice advice = new ReceiveAdvice(); - advice.interceptors.addAll(this.interceptors); + NameMatchMethodPointcutAdvisor sourceAdvisor = new NameMatchMethodPointcutAdvisor( - advice); + new ReceiveAdvice()); sourceAdvisor.addMethodName("receive"); pf.addAdvisor(sourceAdvisor); this.source = (MessageSource) pf.getProxy(); } + private List interceptors() { + return List.copyOf(this.interceptors); + } + public void setRetryTemplate(RetryTemplate retryTemplate) { this.retryTemplate = retryTemplate; } @@ -211,6 +246,7 @@ public boolean poll(MessageHandler handler, ParameterizedTypeReference type) ackCallback = status -> log.warn("No AcknowledgementCallback defined. Status: " + status.name() + " " + message); } + Exception sendFailure = null; try { setAttributesIfNecessary(message); if (this.retryTemplate == null) { @@ -236,13 +272,17 @@ public boolean poll(MessageHandler handler, ParameterizedTypeReference type) } } } + for (ChannelInterceptor interceptor : interceptors()) { + interceptor.postSend(message, DUMMY_CHANNEL, true); + } return true; } catch (MessagingException e) { + sendFailure = e; if (this.retryTemplate == null && !shouldRequeue(e)) { try { this.messagingTemplate.send(this.errorChannel, - this.errorMessageStrategy.buildErrorMessage(e, ATTRIBUTES_HOLDER.get())); + this.errorMessageStrategy.buildErrorMessage(e, ATTRIBUTES_HOLDER.get())); } catch (MessagingException e1) { requeueOrNack(message, ackCallback, e1); @@ -255,6 +295,7 @@ public boolean poll(MessageHandler handler, ParameterizedTypeReference type) } } catch (Exception e) { + sendFailure = e; AckUtils.autoNack(ackCallback); if (e instanceof MessageHandlingException messageHandlingException && messageHandlingException.getFailedMessage().equals(message)) { @@ -265,6 +306,10 @@ public boolean poll(MessageHandler handler, ParameterizedTypeReference type) finally { ATTRIBUTES_HOLDER.remove(); AckUtils.autoAck(ackCallback); + for (ChannelInterceptor interceptor : interceptors()) { + interceptor.afterSendCompletion(message, DUMMY_CHANNEL, + sendFailure == null, sendFailure); + } } } diff --git a/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java new file mode 100644 index 0000000000..fe855c982f --- /dev/null +++ b/core/spring-cloud-stream/src/test/java/org/springframework/cloud/stream/binder/DefaultPollableMessageSourceTests.java @@ -0,0 +1,221 @@ +/* + * Copyright 2026-present 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.cloud.stream.binder; + +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.atomic.AtomicInteger; + +import org.junit.jupiter.api.Test; + +import org.springframework.integration.channel.DirectChannel; +import org.springframework.integration.core.MessageSource; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageChannel; +import org.springframework.messaging.MessageHandler; +import org.springframework.messaging.support.ChannelInterceptor; +import org.springframework.messaging.support.MessageBuilder; + +import static org.assertj.core.api.Assertions.assertThat; + +/** + * Verifies that {@link DefaultPollableMessageSource} invokes the full + * {@link ChannelInterceptor} contract around poll lifecycles (GH-3131), + * while keeping the legacy {@code preSend} behavior intact. + */ +class DefaultPollableMessageSourceTests { + + private final List events = new ArrayList<>(); + + private final AtomicInteger receiveCount = new AtomicInteger(); + + @Test + void successfulPollInvokesFullInterceptorLifecycleInOrder() { + DefaultPollableMessageSource source = newSource(); + source.setSource(this::message); + source.addInterceptor(recorder()); + + assertThat(source.poll(noopHandler())).isTrue(); + + assertThat(this.events).containsExactly( + "preReceive", + "postReceive", + "preSend", + "afterReceiveCompletion", + "postSend", + "afterSendCompletion"); + } + + @Test + void falsePreReceiveShortCircuitsReceive() { + DefaultPollableMessageSource source = newSource(); + source.setSource(this::countingMessage); + source.addInterceptor(new Recorder() { + @Override + public boolean preReceive(MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("preReceive"); + return false; + } + }); + + assertThat(source.poll(noopHandler())).isFalse(); + assertThat(this.events).containsExactly("preReceive"); + assertThat(this.receiveCount.get()).isZero(); + } + + @Test + void nullPostReceiveAbortsFurtherProcessing() { + DefaultPollableMessageSource source = newSource(); + source.setSource(this::countingMessage); + source.addInterceptor(new Recorder() { + @Override + public Message postReceive(Message message, MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("postReceive"); + return null; + } + }); + + assertThat(source.poll(noopHandler())).isFalse(); + assertThat(this.receiveCount.get()).isEqualTo(1); + assertThat(this.events).containsExactly( + "preReceive", + "postReceive", + "afterReceiveCompletion"); + } + + @Test + void handlerFailureSkipsPostSendAndReportsExceptionOnCompletion() { + DefaultPollableMessageSource source = newSource(); + DirectChannel errorChannel = new DirectChannel(); + errorChannel.subscribe(message -> { + }); + source.setErrorChannel(errorChannel); + source.setSource(this::message); + source.addInterceptor(recorder()); + MessageHandler failingHandler = message -> { + throw new IllegalStateException("boom"); + }; + + assertThat(source.poll(failingHandler)).isTrue(); + + assertThat(this.events).containsExactly( + "preReceive", + "postReceive", + "preSend", + "afterReceiveCompletion", + "afterSendCompletion:exception"); + } + + @Test + void interceptorsAddedAfterSetSourceAreHonored() { + DefaultPollableMessageSource source = newSource(); + source.setSource(this::message); + source.addInterceptor(recorder()); + + assertThat(source.poll(noopHandler())).isTrue(); + + assertThat(this.events).contains("postSend", "afterSendCompletion"); + } + + @Test + void nullPreSendStillAbortsLegacyPath() { + DefaultPollableMessageSource source = newSource(); + source.setSource(this::countingMessage); + source.addInterceptor(new Recorder() { + @Override + public Message preSend(Message message, MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("preSend"); + return null; + } + }); + + assertThat(source.poll(noopHandler())).isFalse(); + assertThat(this.receiveCount.get()).isEqualTo(1); + assertThat(this.events).containsExactly( + "preReceive", + "postReceive", + "preSend", + "afterReceiveCompletion"); + } + + private DefaultPollableMessageSource newSource() { + return new DefaultPollableMessageSource(null); + } + + private Message message() { + return MessageBuilder.withPayload((Object) "hello").build(); + } + + private Message countingMessage() { + this.receiveCount.incrementAndGet(); + return message(); + } + + private ChannelInterceptor recorder() { + return new Recorder(); + } + + private MessageHandler noopHandler() { + return message -> { + }; + } + + private class Recorder implements ChannelInterceptor { + + @Override + public boolean preReceive(MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("preReceive"); + return true; + } + + @Override + public Message postReceive(Message message, MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("postReceive"); + return message; + } + + @Override + public void afterReceiveCompletion(Message message, MessageChannel channel, + Exception ex) { + record("afterReceiveCompletion", ex); + } + + @Override + public Message preSend(Message message, MessageChannel channel) { + DefaultPollableMessageSourceTests.this.events.add("preSend"); + return message; + } + + @Override + public void postSend(Message message, MessageChannel channel, boolean sent) { + DefaultPollableMessageSourceTests.this.events.add("postSend"); + } + + @Override + public void afterSendCompletion(Message message, MessageChannel channel, + boolean sent, Exception ex) { + record("afterSendCompletion", ex); + } + + private void record(String name, Exception ex) { + DefaultPollableMessageSourceTests.this.events + .add(ex != null ? name + ":exception" : name); + } + + } + +}