Skip to content

[#2273] activemq-pool support for JMS 3.1/2.0 #2274

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

Open
mattrpav wants to merge 1 commit into
base: main
Choose a base branch
from
Open

[#2273] activemq-pool support for JMS 3.1/2.0 #2274

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
88 changes: 88 additions & 0 deletions activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupport.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You 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
*
* http://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.apache.activemq.jms.pool;

import jakarta.jms.IllegalStateRuntimeException;
import jakarta.jms.InvalidClientIDException;
import jakarta.jms.InvalidClientIDRuntimeException;
import jakarta.jms.InvalidDestinationException;
import jakarta.jms.InvalidDestinationRuntimeException;
import jakarta.jms.InvalidSelectorException;
import jakarta.jms.InvalidSelectorRuntimeException;
import jakarta.jms.JMSException;
import jakarta.jms.JMSRuntimeException;
import jakarta.jms.JMSSecurityException;
import jakarta.jms.JMSSecurityRuntimeException;
import jakarta.jms.MessageFormatException;
import jakarta.jms.MessageFormatRuntimeException;
import jakarta.jms.MessageNotWriteableException;
import jakarta.jms.MessageNotWriteableRuntimeException;
import jakarta.jms.ResourceAllocationException;
import jakarta.jms.ResourceAllocationRuntimeException;
import jakarta.jms.TransactionInProgressException;
import jakarta.jms.TransactionInProgressRuntimeException;
import jakarta.jms.TransactionRolledBackException;
import jakarta.jms.TransactionRolledBackRuntimeException;

/**
* Converts checked {@link JMSException} types into their unchecked JMS 2.0
* counterparts, preserving the specific runtime exception subtype the
* specification defines for each checked type.
*
* activemq-jms-pool depends only on the jakarta.jms API to pool any JMS
* provider. No re-use of activemq-client JMSExceptionSupport.
*/
final class JmsPoolExceptionSupport {

private JmsPoolExceptionSupport() {}

static JMSRuntimeException toRuntimeException(JMSException e) {
var message = e.getMessage();
var errorCode = e.getErrorCode();
if (e instanceof jakarta.jms.IllegalStateException) {
return new IllegalStateRuntimeException(message, errorCode, e);
}
if (e instanceof InvalidClientIDException) {
return new InvalidClientIDRuntimeException(message, errorCode, e);
}
if (e instanceof InvalidDestinationException) {
return new InvalidDestinationRuntimeException(message, errorCode, e);
}
if (e instanceof InvalidSelectorException) {
return new InvalidSelectorRuntimeException(message, errorCode, e);
}
if (e instanceof JMSSecurityException) {
return new JMSSecurityRuntimeException(message, errorCode, e);
}
if (e instanceof MessageFormatException) {
return new MessageFormatRuntimeException(message, errorCode, e);
}
if (e instanceof MessageNotWriteableException) {
return new MessageNotWriteableRuntimeException(message, errorCode, e);
}
if (e instanceof ResourceAllocationException) {
return new ResourceAllocationRuntimeException(message, errorCode, e);
}
if (e instanceof TransactionInProgressException) {
return new TransactionInProgressRuntimeException(message, errorCode, e);
}
if (e instanceof TransactionRolledBackException) {
return new TransactionRolledBackRuntimeException(message, errorCode, e);
}
return new JMSRuntimeException(message, errorCode, e);
}
}
42 changes: 22 additions & 20 deletions activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,7 +91,13 @@ public void close() throws JMSException {

@Override
public void start() throws JMSException {
assertNotClosed();
// Do not use assertNotClosed() here, b/c we need to be able to restart
if (pool == null) {
throw new IllegalStateException("Connection closed");
}
// Connection.stop() only pauses this facade; per JMS a stopped
// connection can be restarted, so clear the stopped flag here.
stopped = false;
pool.start();
}

Expand Down Expand Up @@ -150,8 +156,6 @@ public ConnectionConsumer createConnectionConsumer(Queue queue, String selector,
return getConnection().createConnectionConsumer(queue, selector, serverSessionPool, maxMessages);
}

// Session factory methods
// -------------------------------------------------------------------------
@Override
public QueueSession createQueueSession(boolean transacted, int ackMode) throws JMSException {
return (QueueSession) createSession(transacted, ackMode);
Expand All @@ -173,7 +177,7 @@ public TopicSession createTopicSession(boolean transacted, int ackMode) throws J
*/
@Override
public Session createSession() throws JMSException {
throw new UnsupportedOperationException("createSession() is unsupported");
return createSession(false, Session.AUTO_ACKNOWLEDGE);
}

/**
Expand All @@ -197,15 +201,16 @@ public Session createSession() throws JMSException {
*/
@Override
public Session createSession(int sessionMode) throws JMSException {
throw new UnsupportedOperationException("createSession(int sessionMode) is unsupported");
boolean transacted = sessionMode == Session.SESSION_TRANSACTED;
return createSession(transacted, transacted ? Session.SESSION_TRANSACTED : sessionMode);
}

@Override
public Session createSession(boolean transacted, int ackMode) throws JMSException {
PooledSession result = (PooledSession) pool.createSession(transacted, ackMode);
var result = (PooledSession) pool.createSession(transacted, ackMode);

// Store the session so we can close the sessions that this PooledConnection
// created in order to ensure that consumers etc are closed per the JMS contract.
// created in order to ensure that consumers are closed per the JMS contract.
loanedSessions.add(result);

// Add a event listener to the session that notifies us when the session
Expand All @@ -220,25 +225,22 @@ public Session createSession(boolean transacted, int ackMode) throws JMSExceptio
* @since 2.0
*/
@Override
public ConnectionConsumer createSharedConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool,
public ConnectionConsumer createSharedConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
throw new UnsupportedOperationException("createSharedConnectionConsumer() is not supported");
return getConnection().createSharedConnectionConsumer(topic, subscriptionName, messageSelector, sessionPool, maxMessages);
}

/**
*
*
* @see jakarta.jms.ConnectionConsumer
* @since 2.0
*/
@Override
public ConnectionConsumer createSharedDurableConnectionConsumer(Topic topic, String subscriptionName, String messageSelector, ServerSessionPool sessionPool,
int maxMessages) throws JMSException {
throw new UnsupportedOperationException("createSharedConnectionConsumer() is not supported");
return getConnection().createSharedDurableConnectionConsumer(topic, subscriptionName, messageSelector, sessionPool, maxMessages);
}

// Implementation methods
// -------------------------------------------------------------------------

@Override
public void onTemporaryQueueCreate(TemporaryQueue tempQueue) {
connTempQueues.add(tempQueue);
Expand Down Expand Up @@ -277,16 +279,16 @@ public String toString() {
}

/**
* Remove all of the temporary destinations created for this connection.
* Remove all the temporary destinations created for this connection.
* This is important since the underlying connection may be reused over a
* long period of time, accumulating all of the temporary destinations from
* long period of time, accumulating all the temporary destinations from
* each use. However, from the perspective of the lifecycle from the
* client's view, close() closes the connection and, therefore, deletes all
* of the temporary destinations created.
* the temporary destinations created.
*/
protected void cleanupConnectionTemporaryDestinations() {

for (TemporaryQueue tempQueue : connTempQueues) {
for (var tempQueue : connTempQueues) {
try {
tempQueue.delete();
} catch (JMSException ex) {
Expand All @@ -295,7 +297,7 @@ protected void cleanupConnectionTemporaryDestinations() {
}
connTempQueues.clear();

for (TemporaryTopic tempTopic : connTempTopics) {
for (var tempTopic : connTempTopics) {
try {
tempTopic.delete();
} catch (JMSException ex) {
Expand All @@ -312,7 +314,7 @@ protected void cleanupConnectionTemporaryDestinations() {
*/
protected void cleanupAllLoanedSessions() {

for (PooledSession session : loanedSessions) {
for (var session : loanedSessions) {
try {
session.close();
} catch (JMSException ex) {
Expand Down
74 changes: 59 additions & 15 deletions activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnectionFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,10 +23,12 @@

import jakarta.jms.Connection;
import jakarta.jms.ConnectionFactory;
import jakarta.jms.IllegalStateRuntimeException;
import jakarta.jms.JMSContext;
import jakarta.jms.JMSException;
import jakarta.jms.QueueConnection;
import jakarta.jms.QueueConnectionFactory;
import jakarta.jms.Session;
import jakarta.jms.TopicConnection;
import jakarta.jms.TopicConnectionFactory;

Expand Down Expand Up @@ -89,15 +91,15 @@ public class PooledConnectionFactory implements ConnectionFactory, QueueConnecti

public void initConnectionsPool() {
if (this.connectionsPool == null) {
final GenericKeyedObjectPoolConfig poolConfig = new GenericKeyedObjectPoolConfig();
final var poolConfig = new GenericKeyedObjectPoolConfig();
poolConfig.setJmxEnabled(false);
this.connectionsPool = new GenericKeyedObjectPool<ConnectionKey, ConnectionPool>(
new KeyedPooledObjectFactory<ConnectionKey, ConnectionPool>() {
@Override
public PooledObject<ConnectionPool> makeObject(ConnectionKey connectionKey) throws Exception {
Connection delegate = createConnection(connectionKey);
var delegate = createConnection(connectionKey);

ConnectionPool connection = createConnectionPool(delegate);
var connection = createConnectionPool(delegate);
connection.setIdleTimeout(getIdleTimeout());
connection.setExpiryTimeout(getExpiryTimeout());
connection.setMaximumActiveSessionPerConnection(getMaximumActiveSessionPerConnection());
Expand All @@ -117,7 +119,7 @@ public PooledObject<ConnectionPool> makeObject(ConnectionKey connectionKey) thro

@Override
public void destroyObject(ConnectionKey connectionKey, PooledObject<ConnectionPool> pooledObject) throws Exception {
ConnectionPool connection = pooledObject.getObject();
var connection = pooledObject.getObject();
try {
LOG.trace("Destroying connection: {}", connection);
connection.close();
Expand All @@ -128,7 +130,7 @@ public void destroyObject(ConnectionKey connectionKey, PooledObject<ConnectionPo

@Override
public boolean validateObject(ConnectionKey connectionKey, PooledObject<ConnectionPool> pooledObject) {
ConnectionPool connection = pooledObject.getObject();
var connection = pooledObject.getObject();
if (connection != null && connection.expiredCheck()) {
LOG.trace("Connection has expired: {} and will be destroyed", connection);
return false;
Expand Down Expand Up @@ -218,7 +220,7 @@ public synchronized Connection createConnection(String userName, String password
}

ConnectionPool connection = null;
ConnectionKey key = new ConnectionKey(userName, password);
var key = new ConnectionKey(userName, password);

// This will either return an existing non-expired ConnectionPool or it
// will create a new one to meet the demand.
Expand Down Expand Up @@ -273,43 +275,85 @@ public synchronized Connection createConnection(String userName, String password
}

/**
* @return Returns the JMSContext.
* Creates a {@link JMSContext} backed by a pooled connection, using the default
* user identity and {@link Session#AUTO_ACKNOWLEDGE}. Closing the context
* returns its connection and session to the pool rather than closing them.
*
* @return a JMSContext backed by a pooled connection
* @since 2.0
*/
@Override
public JMSContext createContext() {
throw new UnsupportedOperationException("createContext() is not supported");
return createContext(Session.AUTO_ACKNOWLEDGE);
}

/**
* @return Returns the JMSContext.
* Creates a {@link JMSContext} backed by a pooled connection for the given user
* identity, using {@link Session#AUTO_ACKNOWLEDGE}. Pooled connections are keyed
* by the userName and password, so each distinct identity draws from its own
* pool of connections. Closing the context returns its connection and session
* to the pool rather than closing them.
*
* @return a JMSContext backed by a pooled connection for the given identity
* @since 2.0
*/
@Override
public JMSContext createContext(String userName, String password) {
throw new UnsupportedOperationException("createContext(userName, password) is not supported");
return createContext(userName, password, Session.AUTO_ACKNOWLEDGE);
}

/**
* @return Returns the JMSContext.
* Creates a {@link JMSContext} backed by a pooled connection for the given user
* identity and session mode. Pooled connections are keyed by the userName and
* password, so each distinct identity draws from its own pool of connections.
* Closing the context returns its connection and session to the pool rather
* than closing them.
*
* @return a JMSContext backed by a pooled connection for the given identity
* @throws jakarta.jms.IllegalStateRuntimeException if the factory is stopped
* @since 2.0
*/
@Override
public JMSContext createContext(String userName, String password, int sessionMode) {
throw new UnsupportedOperationException("createContext(userName, password, sessionMode) is not supported");
try {
var pooledConnection = (PooledConnection) createConnection(userName, password);
if (pooledConnection == null) {
throw new IllegalStateRuntimeException("PooledConnectionFactory is stopped");
}
return new PooledJMSContext(pooledConnection, sessionMode);
} catch (JMSException e) {
throw JmsPoolExceptionSupport.toRuntimeException(e);
}
}

/**
* @return Returns the JMSContext.
* Creates a {@link JMSContext} backed by a pooled connection, using the default
* user identity and the given session mode. Closing the context returns its
* connection and session to the pool rather than closing them.
*
* @return a JMSContext backed by a pooled connection
* @throws jakarta.jms.IllegalStateRuntimeException if the factory is stopped
* @since 2.0
*/
@Override
public JMSContext createContext(int sessionMode) {
throw new UnsupportedOperationException("createContext(sessionMode) is not supported");
try {
var pooledConnection = (PooledConnection) createConnection();
if (pooledConnection == null) {
throw new IllegalStateRuntimeException("PooledConnectionFactory is stopped");
}
return new PooledJMSContext(pooledConnection, sessionMode);
} catch (JMSException e) {
throw JmsPoolExceptionSupport.toRuntimeException(e);
}
}

protected Connection newPooledConnection(ConnectionPool connection) {
return new PooledConnection(connection);
}

private JMSException createJmsException(String msg, Exception cause) {
JMSException exception = new JMSException(msg);
var exception = new JMSException(msg);
exception.setLinkedException(cause);
exception.initCause(cause);
return exception;
Expand Down
Loading
Loading