Skip to content

GH-10090: Add AmqpClientInboundGateway #10447

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom
Merged
merged 4 commits into from
Oct 6, 2025
Merged

GH-10090: Add AmqpClientInboundGateway #10447

merged 4 commits into from
Oct 6, 2025

Conversation

@artembilan
Copy link
Member

Related to: #10090

  • Add AmqpClientInboundGateway that is mostly a copy/paste of the AmqpClientMessageProducer, but adds a reply-producing logic
  • Cover with tests and document this new component
  • Fix a couple typos in the amqp-1.0.adoc

@artembilan artembilan added this to the 7.0.0-RC1 milestone Sep 26, 2025
@artembilan artembilan requested a review from cppwfs September 26, 2025 19:42
@artembilan
Copy link
Member Author

After doing this, I don't see a value in the AmqpClientMessageSource: who would need to poll messages from the queue on demand? 🤷

The RabbitAmqpTemplate.receive() API could be turned into a MessageSource very easy:

@Bean 
IntegrationFlow amqpClientPollingFlow(RabbitAmqpTemplate rabbitAmqpTemplate) {
    return IntegrationFlow.fromSupplier(rabbitAmqpTemplate::receive)
                          .handle((p, h) -> p, e -> e.async(true))
                          ...
}

So, the point is that RabbitAmqpTemplate.receive() returns a CompletableFuture<Message> which we don't want to join() immediately.

cppwfs
cppwfs requested changes Sep 26, 2025
View reviewed changes
Copy link
Contributor

@cppwfs cppwfs left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looks great!
Just a couple of doc questions

@artembilan artembilan requested a review from cppwfs September 26, 2025 20:26
cppwfs
cppwfs approved these changes Sep 26, 2025
View reviewed changes
Copy link
Contributor

@cppwfs cppwfs left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sweet!

I'll merge once it successfully builds.

@artembilan
Copy link
Member Author

Hold on from merging, please.

@cppwfs
Copy link
Contributor

cppwfs commented Sep 26, 2025

On hold till I get the go ahead.

@artembilan artembilan requested a review from cppwfs September 26, 2025 21:22
cppwfs
cppwfs requested changes Oct 6, 2025
View reviewed changes
Copy link
Contributor

@cppwfs cppwfs left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Great job on consolidating the code. The red is always pleasing to see! :-)

Just a couple of nitpicks.

...qp/src/main/java/org/springframework/integration/amqp/inbound/AmqpClientMessageProducer.java Outdated
* @param delegate the {@link AmqpAcknowledgment} to delegate to.
* Use as {@link java.util.function.BiConsumer} for the {@link IntegrationRabbitAmqpMessageListener}.
* @param messageToSend the message to produce from this endpoint.
* @param requestMessage the request AMQP message.
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

may want to mention that the requestMessage is ignored in this processRequest.

.../java/org/springframework/integration/amqp/inbound/IntegrationRabbitAmqpMessageListener.java
try {
this.requestAction.accept(messageToSend, message);
}
catch (Exception ex) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Shouldn't this be a RuntimeException?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That's the point.
We have to re-throw anything as a ListenerExecutionFailedException and ErrorHandler in the listener container will be able to parse this and its cause respectively.

.../java/org/springframework/integration/amqp/inbound/IntegrationRabbitAmqpMessageListener.java Outdated

@Override
public void onMessage(org.springframework.amqp.core.Message message) {
throw new UnsupportedOperationException("The 'RabbitAmqpMessageListener' does not implement 'onMessage()'");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should it read, The 'IntegrationRabbitAmqpMessageListener' does not implement 'onMessge'?

@artembilan
spring-projectsGH-10090: Add AmqpClientInboundGateway
fef21b1 
Related to: spring-projects#10090

* Add `AmqpClientInboundGateway` that is mostly a copy/paste of the `AmqpClientMessageProducer`,
but adds a reply-producing logic
* Cover with tests and document this new component
* Fix a couple typos in the `amqp-1.0.adoc`
@artembilan
Fix docs typos
10839a7
@artembilan
Extract the IntegrationRabbitAmqpMessageListener as a top-level (pa…
c05adbb 
…ckage private) class.

Serves as a reusable unit of work for both `AmqpClientMessageProducer` & `AmqpClientInboundGateway`.
The one-way and request-reply parts are handled as a `BiConsumer` action injection
into this `IntegrationRabbitAmqpMessageListener` instance.
Now both `AmqpClientMessageProducer` & `AmqpClientInboundGateway` are much simpler.
@artembilan
* Improve AmqpClientMessageProducer.processRequest Javadoc
73cf36c 
mentioning that `requestMessage` param is out of use
* Be more specific with the type for `UnsupportedOperationException`
in the `IntegrationRabbitAmqpMessageListener`
@artembilan artembilan force-pushed the GH-10090-inbound-gateway branch from fe22890 to 73cf36c Compare October 6, 2025 17:29
cppwfs
cppwfs approved these changes Oct 6, 2025
View reviewed changes
Copy link
Contributor

@cppwfs cppwfs left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

LGTM!

This is Awesome!

@cppwfs cppwfs merged commit aad2f7b into spring-projects:main Oct 6, 2025
3 checks passed
@artembilan artembilan deleted the GH-10090-inbound-gateway branch October 6, 2025 20:29
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@cppwfs cppwfs cppwfs approved these changes

Assignees

No one assigned

Labels

in: amqp type: enhancement

Projects

None yet

Milestone

7.0.0-RC1

Development

Successfully merging this pull request may close these issues.

2 participants

@artembilan @cppwfs