Skip to content

GH-3131 Invoke full ChannelInterceptor contract in DefaultPollableMessageSource - #3253

Open
akenra wants to merge 1 commit into
spring-cloud:mainfrom
akenra:feat/gh-3131
Open

GH-3131 Invoke full ChannelInterceptor contract in DefaultPollableMessageSource#3253
akenra wants to merge 1 commit into
spring-cloud:mainfrom
akenra:feat/gh-3131

Conversation

@akenra

@akenra akenra commented Aug 24, 2026

Copy link
Copy Markdown
Contributor

Context

DefaultPollableMessageSource currently invokes only the preSend method of registered ChannelInterceptors (via an AOP advice around MessageSource.receive()). As discussed in #3131, @artembilan agreed to implement the rest of the ChannelInterceptor contract on pollable sources while retaining preSend for backward compatibility ("it won't hurt to implement the rest of ChannelInterceptor contracts in the PollableMessageSource").

What this PR implements

Receive phase (around MessageSource.receive()):

  • preReceive(channel) gate before retrieval: if any interceptor returns false, no receive is attempted and (per its javadoc precondition) afterReceiveCompletion is not invoked.
  • postReceive(message, channel) after retrieval: may transform the message; returning null drops it.
  • afterReceiveCompletion(message, channel, ex) in a finally block once the preReceive chain has passed.

Handling phase (in poll()):

  • postSend(message, channel, true) only after successful handling.
  • afterSendCompletion(message, channel, sent, ex) exactly once in finally, where sent is true only when no failure was recorded; handled-error paths (error channel) record the exception while still returning true from poll().

Lifecycle order for a successful poll:

preReceive -> postReceive -> preSend (legacy, unchanged position)
    -> [receive returns] -> afterReceiveCompletion
    -> [handler] -> postSend -> afterSendCompletion

Bonus fix: setSource() used to snapshot the interceptor list at wiring time, silently ignoring interceptors added afterwards. The advice now reads a live view.

Backward compatibility

  • Legacy preSend keeps its exact position and semantics; all 14 tests of the existing interceptor-driven PollableConsumerTests pass unmodified.
  • No framework-internal code registers ChannelInterceptors on pollable sources (verified), so the affected surface is user interceptors only.

Testing

  • New DefaultPollableMessageSourceTests (6 tests): exact lifecycle ordering, preReceive=false short-circuit, postReceive=null drop, handled-error completion with exception marker and skipped postSend, late addInterceptor() now honored, legacy null-preSend abort preserved. All went red before the fix and green after.
  • Regression: core/spring-cloud-stream full suite 40 tests / 0 failures; PollableConsumerTests 14/14 and PollableSourceTests 5/5.

Resolves #3131

Notes for reviewers

  • postReceive returning null is treated as dropping the message entirely (mirroring pollable-channel behavior). If the preferred semantics are skip-only-that-interceptor, happy to adjust.
  • afterSendCompletion's sent flag is true iff no failure was recorded - on handled-error paths poll() returns true but the hook still sees the exception.
  • Possible follow-ups: multi-interceptor ordering test, RecoveryCallback path test, and the longer-term idea from DefaultPollableMessageSource should execute postSend and afterSendCompletion interceptor methods #3131 of a dedicated non-ChannelInterceptor abstraction for pollables.

…tPollableMessageSource

Signed-off-by: akenra <37288280+akenra@users.noreply.github.com>
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

DefaultPollableMessageSource should execute postSend and afterSendCompletion interceptor methods

1 participant