gnodet-bot commented on code in PR #29336:
URL: https://github.com/apache/camel/pull/29336#discussion_r4493361558
##########
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:
⚠️ **Missing `doShutdown()` → `destroy()`**
`resetConnection()` here is correct for stop/restart — the shared connection
is dropped and re-opens on the next send. But `JmsComponent` has a
`doShutdown()` lifecycle hook (called on final context shutdown) which
`AMQPComponent` doesn't override, so `cachedConnectionFactory.destroy()` is
never called. `destroy()` marks the factory permanently inactive and releases
any pooled resources; without it, the factory stays "active" after shutdown.
```suggestion
@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();
}
@Override
protected void doShutdown() throws Exception {
if (cachedConnectionFactory != null) {
cachedConnectionFactory.destroy();
cachedConnectionFactory = null;
}
super.doShutdown();
}
```
--
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]
