davsclaus commented on code in PR #29336:
URL: https://github.com/apache/camel/pull/29336#discussion_r4493784607
##########
components/camel-amqp/src/main/java/org/apache/camel/component/amqp/AMQPComponent.java:
##########
@@ -156,18 +178,67 @@ protected void doInit() throws Exception {
connectionFactory.setTopicPrefix("topic://");
}
getConfiguration().setConnectionFactory(connectionFactory);
+ createdConnectionFactory = connectionFactory;
}
}
+ if (cacheConnection && createdConnectionFactory != null
+ && createdConnectionFactory ==
getConfiguration().getConnectionFactory()
+ && getConfiguration() instanceof AMQPConfiguration
configuration) {
+ cachedConnectionFactory =
createCachedConnectionFactory(createdConnectionFactory);
+ configuration.useCachedConnection(createdConnectionFactory,
cachedConnectionFactory);
+ }
super.doInit();
}
+ private CachingConnectionFactory
createCachedConnectionFactory(ConnectionFactory connectionFactory) {
+ ConnectionFactory target = connectionFactory;
+ if (getUsername() != null && getPassword() != null) {
+ // as the endpoints do, see JmsComponent#createEndpoint
+ UserCredentialsConnectionFactoryAdapter adapter = new
UserCredentialsConnectionFactoryAdapter();
+ adapter.setTargetConnectionFactory(connectionFactory);
+ adapter.setUsername(getUsername());
+ adapter.setPassword(getPassword());
+ target = adapter;
+ }
+ CachingConnectionFactory answer = new CachingConnectionFactory(target);
+ // a polling consumer (such as pollEnrich) closes its consumer after
each receive, so it does not hold on to
+ // the messages that the broker sends it ahead
+ answer.setCacheConsumers(false);
+ // a producer is created for each message, so the next message does
not use a producer the broker has closed
+ answer.setCacheProducers(false);
+ return answer;
+ }
+
@Override
protected Endpoint createEndpoint(String uri, String remaining,
Map<String, Object> parameters) throws Exception {
+ boolean endpointConnectionFactory =
parameters.containsKey("connectionFactory");
JmsEndpoint endpoint = (JmsEndpoint) super.createEndpoint(uri,
remaining, parameters);
endpoint.setBinding(new AMQPJmsBinding(endpoint));
+ if (cachedConnectionFactory != null && endpoint.getConfiguration()
instanceof AMQPConfiguration configuration) {
+ // the endpoint uses the connection factory that the component
created, which JmsComponent#createEndpoint
+ // wraps with the credentials of the component when they are set
+ boolean createdConnectionFactoryUsed = !endpointConnectionFactory
+ && getConfiguration().getConnectionFactory() ==
createdConnectionFactory
+ && Objects.equals(configuration.getUsername(),
getUsername())
+ && Objects.equals(configuration.getPassword(),
getPassword());
+ if (createdConnectionFactoryUsed) {
+
configuration.useCachedConnection(configuration.getConnectionFactory(),
cachedConnectionFactory);
+ } else {
+ configuration.useCachedConnection(null, null);
+ }
+ }
return endpoint;
}
+ @Override
+ protected void doStop() throws Exception {
+ if (cachedConnectionFactory != null) {
+ // closes the shared connection, which opens again on the next
message sent
+ cachedConnectionFactory.resetConnection();
+ }
+ super.doStop();
+ }
Review Comment:
Checked against Spring 7.0.x: `SingleConnectionFactory.destroy()` only calls
`resetConnection()`, which `doStop()` already does, so no `doShutdown()`
override is needed (as the author explained in the PR comments). Resolving.
_Claude Code on behalf of davsclaus_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
