diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupport.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupport.java new file mode 100644 index 00000000000..c9e994a0f99 --- /dev/null +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupport.java @@ -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); + } +} diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnection.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnection.java index f194c1e6b26..70ccc056f45 100644 --- a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnection.java +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnection.java @@ -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(); } @@ -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); @@ -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); } /** @@ -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 @@ -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); @@ -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) { @@ -295,7 +297,7 @@ protected void cleanupConnectionTemporaryDestinations() { } connTempQueues.clear(); - for (TemporaryTopic tempTopic : connTempTopics) { + for (var tempTopic : connTempTopics) { try { tempTopic.delete(); } catch (JMSException ex) { @@ -312,7 +314,7 @@ protected void cleanupConnectionTemporaryDestinations() { */ protected void cleanupAllLoanedSessions() { - for (PooledSession session : loanedSessions) { + for (var session : loanedSessions) { try { session.close(); } catch (JMSException ex) { diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnectionFactory.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnectionFactory.java index d3a64e1934e..7a8953a0407 100644 --- a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnectionFactory.java +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledConnectionFactory.java @@ -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; @@ -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( new KeyedPooledObjectFactory() { @Override public PooledObject 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()); @@ -117,7 +119,7 @@ public PooledObject makeObject(ConnectionKey connectionKey) thro @Override public void destroyObject(ConnectionKey connectionKey, PooledObject pooledObject) throws Exception { - ConnectionPool connection = pooledObject.getObject(); + var connection = pooledObject.getObject(); try { LOG.trace("Destroying connection: {}", connection); connection.close(); @@ -128,7 +130,7 @@ public void destroyObject(ConnectionKey connectionKey, PooledObject 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; @@ -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. @@ -273,35 +275,77 @@ 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) { @@ -309,7 +353,7 @@ protected Connection newPooledConnection(ConnectionPool 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; diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSConsumer.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSConsumer.java new file mode 100644 index 00000000000..3878ab601e2 --- /dev/null +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSConsumer.java @@ -0,0 +1,148 @@ +/** + * 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.JMSConsumer; +import jakarta.jms.JMSException; +import jakarta.jms.JMSRuntimeException; +import jakarta.jms.Message; +import jakarta.jms.MessageConsumer; +import jakarta.jms.MessageListener; + +/** + * JMS 2.0 {@link JMSConsumer} backed by a pooled {@link MessageConsumer}. + */ +public class PooledJMSConsumer implements JMSConsumer { + + private final PooledJMSContext context; + private final MessageConsumer consumer; + + PooledJMSConsumer(PooledJMSContext context, MessageConsumer consumer) { + this.context = context; + this.consumer = consumer; + } + + @Override + public String getMessageSelector() { + try { + return consumer.getMessageSelector(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public MessageListener getMessageListener() throws JMSRuntimeException { + try { + return consumer.getMessageListener(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void setMessageListener(MessageListener listener) throws JMSRuntimeException { + try { + consumer.setMessageListener(listener); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Message receive() { + try { + return trackReceived(consumer.receive()); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Message receive(long timeout) { + try { + return trackReceived(consumer.receive(timeout)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Message receiveNoWait() { + try { + return trackReceived(consumer.receiveNoWait()); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + private Message trackReceived(Message message) { + if (message != null && context != null) { + context.onMessageReceived(message); + } + return message; + } + + @Override + public T receiveBody(Class c) { + var message = receive(); + if (message == null) { + return null; + } + try { + return message.getBody(c); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public T receiveBody(Class c, long timeout) { + var message = receive(timeout); + if (message == null) { + return null; + } + try { + return message.getBody(c); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public T receiveBodyNoWait(Class c) { + var message = receiveNoWait(); + if (message == null) { + return null; + } + try { + return message.getBody(c); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void close() { + try { + consumer.close(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + +} diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSContext.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSContext.java new file mode 100644 index 00000000000..0eaaa72dbdb --- /dev/null +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSContext.java @@ -0,0 +1,627 @@ +/** + * 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 java.io.Serializable; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; + +import jakarta.jms.BytesMessage; +import jakarta.jms.ConnectionMetaData; +import jakarta.jms.Destination; +import jakarta.jms.ExceptionListener; +import jakarta.jms.IllegalStateRuntimeException; +import jakarta.jms.JMSConsumer; +import jakarta.jms.JMSContext; +import jakarta.jms.JMSException; +import jakarta.jms.JMSProducer; +import jakarta.jms.JMSRuntimeException; +import jakarta.jms.MapMessage; +import jakarta.jms.Message; +import jakarta.jms.MessageProducer; +import jakarta.jms.ObjectMessage; +import jakarta.jms.Queue; +import jakarta.jms.QueueBrowser; +import jakarta.jms.Session; +import jakarta.jms.StreamMessage; +import jakarta.jms.TemporaryQueue; +import jakarta.jms.TemporaryTopic; +import jakarta.jms.TextMessage; +import jakarta.jms.Topic; + +/** + * JMS 2.0 {@link JMSContext} backed by a {@link PooledConnection} and a + * lazily-created {@link Session}. Child contexts created via + * {@link #createContext(int)} share the same underlying connection and use + * an {@link AtomicLong} reference counter so the connection is returned to + * the pool only when the last context closes. + */ +public class PooledJMSContext implements JMSContext { + + private final PooledConnection connection; + private final int sessionMode; + private final AtomicLong connectionCounter; + private final AtomicBoolean closed = new AtomicBoolean(); + private volatile boolean autoStart = true; + private volatile Session session; + private volatile MessageProducer contextProducer; + private volatile Message lastReceivedMessage; + + PooledJMSContext(PooledConnection connection, int sessionMode) { + this(connection, sessionMode, new AtomicLong(1)); + } + + private PooledJMSContext(PooledConnection connection, int sessionMode, AtomicLong connectionCounter) { + this.connection = connection; + this.sessionMode = sessionMode; + this.connectionCounter = connectionCounter; + } + + @Override + public JMSContext createContext(int sessionMode) { + checkClosed(); + // Increment-if-positive so a concurrent last-reference close() cannot + // release the connection while we attach a new context to it. + long current; + do { + current = connectionCounter.get(); + if (current <= 0) { + throw new IllegalStateRuntimeException("Context is closed"); + } + } while (!connectionCounter.compareAndSet(current, current + 1)); + return new PooledJMSContext(connection, sessionMode, connectionCounter); + } + + @Override + public JMSProducer createProducer() { + checkClosed(); + try { + var s = getSession(); + var producer = contextProducer; + if (producer == null) { + synchronized (this) { + checkClosed(); + if (contextProducer == null) { + contextProducer = s.createProducer(null); + } + producer = contextProducer; + } + } + return new PooledJMSProducer(s, producer); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public String getClientID() { + checkClosed(); + try { + return connection.getClientID(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void setClientID(String clientID) { + checkClosed(); + try { + connection.setClientID(clientID); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public ConnectionMetaData getMetaData() { + checkClosed(); + try { + return connection.getMetaData(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public ExceptionListener getExceptionListener() { + checkClosed(); + try { + return connection.getExceptionListener(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void setExceptionListener(ExceptionListener listener) { + checkClosed(); + try { + connection.setExceptionListener(listener); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void start() { + checkClosed(); + try { + connection.start(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void stop() { + checkClosed(); + try { + connection.stop(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void setAutoStart(boolean autoStart) { + checkClosed(); + this.autoStart = autoStart; + } + + @Override + public boolean getAutoStart() { + checkClosed(); + return autoStart; + } + + @Override + public void close() { + if (!closed.compareAndSet(false, true)) { + return; + } + JMSRuntimeException failure = null; + try { + synchronized (this) { + var producerFailure = resetContextProducer(); + if (producerFailure != null) { + failure = JmsPoolExceptionSupport.toRuntimeException(producerFailure); + } + if (session != null) { + try { + session.close(); + } catch (JMSException e) { + if (failure == null) { + failure = JmsPoolExceptionSupport.toRuntimeException(e); + } + } finally { + session = null; + } + } + } + } finally { + lastReceivedMessage = null; + if (connectionCounter.decrementAndGet() == 0) { + try { + connection.close(); + } catch (JMSException e) { + if (failure == null) { + failure = JmsPoolExceptionSupport.toRuntimeException(e); + } + } + } + } + if (failure != null) { + throw failure; + } + } + + @Override + public BytesMessage createBytesMessage() { + checkClosed(); + try { + return getSession().createBytesMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public MapMessage createMapMessage() { + checkClosed(); + try { + return getSession().createMapMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Message createMessage() { + checkClosed(); + try { + return getSession().createMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public ObjectMessage createObjectMessage() { + checkClosed(); + try { + return getSession().createObjectMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public ObjectMessage createObjectMessage(Serializable object) { + checkClosed(); + try { + return getSession().createObjectMessage(object); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public StreamMessage createStreamMessage() { + checkClosed(); + try { + return getSession().createStreamMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public TextMessage createTextMessage() { + checkClosed(); + try { + return getSession().createTextMessage(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public TextMessage createTextMessage(String text) { + checkClosed(); + try { + return getSession().createTextMessage(text); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public boolean getTransacted() { + checkClosed(); + return sessionMode == Session.SESSION_TRANSACTED; + } + + @Override + public int getSessionMode() { + checkClosed(); + return sessionMode; + } + + @Override + public void commit() { + checkClosed(); + try { + getSession().commit(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void rollback() { + checkClosed(); + try { + getSession().rollback(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void recover() { + checkClosed(); + try { + getSession().recover(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createConsumer(Destination destination) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createConsumer(destination)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createConsumer(Destination destination, String messageSelector) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createConsumer(destination, messageSelector)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createConsumer(Destination destination, String messageSelector, boolean noLocal) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createConsumer(destination, messageSelector, noLocal)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Queue createQueue(String queueName) { + checkClosed(); + try { + return getSession().createQueue(queueName); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public Topic createTopic(String topicName) { + checkClosed(); + try { + return getSession().createTopic(topicName); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createDurableConsumer(Topic topic, String name) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createDurableConsumer(topic, name)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createDurableConsumer(topic, name, messageSelector, noLocal)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createSharedDurableConsumer(Topic topic, String name) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createSharedDurableConsumer(topic, name)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createSharedDurableConsumer(Topic topic, String name, String messageSelector) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createSharedDurableConsumer(topic, name, messageSelector)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createSharedConsumer(topic, sharedSubscriptionName)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public JMSConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName, String messageSelector) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return new PooledJMSConsumer(this, getSession().createSharedConsumer(topic, sharedSubscriptionName, messageSelector)); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public QueueBrowser createBrowser(Queue queue) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return getSession().createBrowser(queue); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public QueueBrowser createBrowser(Queue queue, String messageSelector) { + checkClosed(); + try { + if (autoStart) { + connection.start(); + } + return getSession().createBrowser(queue, messageSelector); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public TemporaryQueue createTemporaryQueue() { + checkClosed(); + try { + return getSession().createTemporaryQueue(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public TemporaryTopic createTemporaryTopic() { + checkClosed(); + try { + return getSession().createTemporaryTopic(); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void unsubscribe(String name) { + checkClosed(); + try { + getSession().unsubscribe(name); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + @Override + public void acknowledge() { + checkClosed(); + try { + // Acknowledging any consumed message acknowledges all messages the + // session has consumed; track the most recent one received through + // this context's consumers. Messages delivered to MessageListeners + // are not tracked (see the module's known-limitations doc). + var last = lastReceivedMessage; + if (last != null) { + last.acknowledge(); + } + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + + void onMessageReceived(Message message) { + lastReceivedMessage = message; + } + + private Session getSession() { + var result = session; + if (result == null || isStale(result)) { + synchronized (this) { + checkClosed(); + if (session != null && isStale(session)) { + // The pooled session was returned behind our back -- typically an + // XA transaction completed and its Synchronization closed the + // session. Drop the stale references and renew so the next + // operation enlists in the current transaction. + resetContextProducer(); + session = null; + } + if (session == null) { + try { + var transacted = sessionMode == Session.SESSION_TRANSACTED; + session = connection.createSession(transacted, + transacted ? Session.SESSION_TRANSACTED : sessionMode); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + } + result = session; + } + } + return result; + } + + private static boolean isStale(Session session) { + return session instanceof PooledSession && ((PooledSession) session).isClosed(); + } + + /** + * Drops the cached producer, closing the underlying MessageProducer when it + * was created for this context (non-anonymous mode); in anonymous mode the + * producer is owned by the pooled session and must survive. Callers must + * hold the context monitor; any close failure is returned for the caller + * to handle. + */ + private JMSException resetContextProducer() { + var producer = contextProducer; + contextProducer = null; + if (producer instanceof PooledProducer && session instanceof PooledSession + && !((PooledSession) session).isUseAnonymousProducers()) { + try { + ((PooledProducer) producer).getMessageProducer().close(); + } catch (JMSException e) { + return e; + } + } + return null; + } + + private void checkClosed() { + if (closed.get()) { + throw new IllegalStateRuntimeException("Context is closed"); + } + } +} diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSProducer.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSProducer.java new file mode 100644 index 00000000000..60934951984 --- /dev/null +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledJMSProducer.java @@ -0,0 +1,511 @@ +/** + * 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 java.io.Serializable; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; +import java.util.Set; + +import jakarta.jms.CompletionListener; +import jakarta.jms.DeliveryMode; +import jakarta.jms.Destination; +import jakarta.jms.IllegalStateRuntimeException; +import jakarta.jms.JMSException; +import jakarta.jms.JMSProducer; +import jakarta.jms.JMSRuntimeException; +import jakarta.jms.Message; +import jakarta.jms.MessageFormatRuntimeException; +import jakarta.jms.MessageProducer; +import jakarta.jms.Session; + +/** + * JMS 2.0 {@link JMSProducer} backed by a pooled {@link Session} and its + * anonymous {@link MessageProducer}. QoS overrides and message properties + * are accumulated locally and applied at send time. + */ +public class PooledJMSProducer implements JMSProducer { + + private final Session session; + private final MessageProducer producer; + + private int deliveryMode = DeliveryMode.PERSISTENT; + private int priority = Message.DEFAULT_PRIORITY; + private long timeToLive = Message.DEFAULT_TIME_TO_LIVE; + private long deliveryDelay = 0; + private boolean disableMessageID = false; + private boolean disableMessageTimestamp = false; + private CompletionListener completionListener; + + private String jmsCorrelationID; + private byte[] jmsCorrelationIDBytes; + private String jmsType; + private Destination jmsReplyTo; + + private Map properties; + + PooledJMSProducer(Session session, MessageProducer producer) throws JMSException { + this.session = session; + this.producer = producer; + this.deliveryMode = producer.getDeliveryMode(); + this.priority = producer.getPriority(); + this.timeToLive = producer.getTimeToLive(); + } + + @Override + public JMSProducer send(Destination destination, Message message) { + checkSessionOpen(); + if (message == null) { + throw new MessageFormatRuntimeException("Message must not be null"); + } + try { + applyHeaders(message); + applyProperties(message); + if (completionListener != null) { + // Pass the listener through; the underlying provider either sends + // asynchronously or rejects the call if it lacks support. + producer.send(destination, message, deliveryMode, priority, timeToLive, completionListener); + } else { + producer.send(destination, message, deliveryMode, priority, timeToLive); + } + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + return this; + } + + @Override + public JMSProducer send(Destination destination, String body) { + checkSessionOpen(); + try { + var message = session.createTextMessage(body); + send(destination, message); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + return this; + } + + @Override + public JMSProducer send(Destination destination, Map body) { + checkSessionOpen(); + try { + var message = session.createMapMessage(); + if (body != null) { + for (var entry : body.entrySet()) { + message.setObject(entry.getKey(), entry.getValue()); + } + } + send(destination, message); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + return this; + } + + @Override + public JMSProducer send(Destination destination, byte[] body) { + checkSessionOpen(); + try { + var message = session.createBytesMessage(); + if (body != null) { + message.writeBytes(body); + } + send(destination, message); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + return this; + } + + @Override + public JMSProducer send(Destination destination, Serializable body) { + checkSessionOpen(); + try { + var message = session.createObjectMessage(body); + send(destination, message); + } catch (JMSException e) { + throw JmsPoolExceptionSupport.toRuntimeException(e); + } + return this; + } + + @Override + public JMSProducer setDeliveryMode(int deliveryMode) { + if (deliveryMode != DeliveryMode.PERSISTENT && deliveryMode != DeliveryMode.NON_PERSISTENT) { + throw new JMSRuntimeException("Unknown delivery mode: " + deliveryMode); + } + this.deliveryMode = deliveryMode; + return this; + } + + @Override + public int getDeliveryMode() { + return deliveryMode; + } + + @Override + public JMSProducer setPriority(int priority) { + if (priority < 0 || priority > 9) { + throw new JMSRuntimeException("Priority must be between 0 and 9"); + } + this.priority = priority; + return this; + } + + @Override + public int getPriority() { + return priority; + } + + @Override + public JMSProducer setTimeToLive(long timeToLive) { + this.timeToLive = timeToLive; + return this; + } + + @Override + public long getTimeToLive() { + return timeToLive; + } + + @Override + public JMSProducer setDeliveryDelay(long deliveryDelay) { + this.deliveryDelay = deliveryDelay; + return this; + } + + @Override + public long getDeliveryDelay() { + return deliveryDelay; + } + + @Override + public JMSProducer setDisableMessageID(boolean value) { + this.disableMessageID = value; + return this; + } + + @Override + public boolean getDisableMessageID() { + return disableMessageID; + } + + @Override + public JMSProducer setDisableMessageTimestamp(boolean value) { + this.disableMessageTimestamp = value; + return this; + } + + @Override + public boolean getDisableMessageTimestamp() { + return disableMessageTimestamp; + } + + @Override + public JMSProducer setAsync(CompletionListener completionListener) { + this.completionListener = completionListener; + return this; + } + + @Override + public CompletionListener getAsync() { + return completionListener; + } + + @Override + public JMSProducer setJMSCorrelationID(String correlationID) { + this.jmsCorrelationID = correlationID; + return this; + } + + @Override + public String getJMSCorrelationID() { + return jmsCorrelationID; + } + + @Override + public JMSProducer setJMSCorrelationIDAsBytes(byte[] correlationID) { + this.jmsCorrelationIDBytes = correlationID != null ? correlationID.clone() : null; + return this; + } + + @Override + public byte[] getJMSCorrelationIDAsBytes() { + return jmsCorrelationIDBytes; + } + + @Override + public JMSProducer setJMSType(String type) { + this.jmsType = type; + return this; + } + + @Override + public String getJMSType() { + return jmsType; + } + + @Override + public JMSProducer setJMSReplyTo(Destination replyTo) { + this.jmsReplyTo = replyTo; + return this; + } + + @Override + public Destination getJMSReplyTo() { + return jmsReplyTo; + } + + @Override + public JMSProducer setProperty(String name, boolean value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, byte value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, short value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, int value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, long value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, float value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, double value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, String value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer setProperty(String name, Object value) { + checkPropertyName(name); + getOrCreateProperties().put(name, value); + return this; + } + + @Override + public JMSProducer clearProperties() { + if (properties != null) { + properties.clear(); + } + return this; + } + + @Override + public boolean propertyExists(String name) { + return properties != null && properties.containsKey(name); + } + + @Override + public boolean getBooleanProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Boolean.valueOf((String) value); + } + if (value instanceof Boolean) { + return (Boolean) value; + } + throw invalidConversion(name, value, "boolean"); + } + + @Override + public byte getByteProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Byte.valueOf((String) value); + } + if (value instanceof Byte) { + return (Byte) value; + } + throw invalidConversion(name, value, "byte"); + } + + @Override + public short getShortProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Short.valueOf((String) value); + } + if (value instanceof Byte || value instanceof Short) { + return ((Number) value).shortValue(); + } + throw invalidConversion(name, value, "short"); + } + + @Override + public int getIntProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Integer.valueOf((String) value); + } + if (value instanceof Byte || value instanceof Short || value instanceof Integer) { + return ((Number) value).intValue(); + } + throw invalidConversion(name, value, "int"); + } + + @Override + public long getLongProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Long.valueOf((String) value); + } + if (value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long) { + return ((Number) value).longValue(); + } + throw invalidConversion(name, value, "long"); + } + + @Override + public float getFloatProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Float.valueOf((String) value); + } + if (value instanceof Float) { + return (Float) value; + } + throw invalidConversion(name, value, "float"); + } + + @Override + public double getDoubleProperty(String name) { + var value = getProperty(name); + if (value == null || value instanceof String) { + return Double.valueOf((String) value); + } + if (value instanceof Float || value instanceof Double) { + return ((Number) value).doubleValue(); + } + throw invalidConversion(name, value, "double"); + } + + @Override + public String getStringProperty(String name) { + var value = getProperty(name); + return value != null ? value.toString() : null; + } + + @Override + public Object getObjectProperty(String name) { + return getProperty(name); + } + + @Override + public Set getPropertyNames() { + if (properties == null) { + return Collections.emptySet(); + } + return properties.keySet(); + } + + /** + * Rejects sends once the backing pooled session has been returned to the + * pool (context closed, or an XA transaction completed and renewed the + * context's session). Without this check send with a pre-built message + * would bypass the session and silently use the pooled anonymous producer, + * which may already be loaned to another borrower. + */ + private void checkSessionOpen() { + if (session instanceof PooledSession && ((PooledSession) session).isClosed()) { + throw new IllegalStateRuntimeException("The producer's pooled session is closed"); + } + } + + private void applyHeaders(Message message) throws JMSException { + if (jmsCorrelationID != null) { + message.setJMSCorrelationID(jmsCorrelationID); + } + if (jmsCorrelationIDBytes != null) { + message.setJMSCorrelationIDAsBytes(jmsCorrelationIDBytes); + } + if (jmsType != null) { + message.setJMSType(jmsType); + } + if (jmsReplyTo != null) { + message.setJMSReplyTo(jmsReplyTo); + } + } + + private void applyProperties(Message message) throws JMSException { + if (properties != null) { + for (var entry : properties.entrySet()) { + message.setObjectProperty(entry.getKey(), entry.getValue()); + } + } + } + + private Map getOrCreateProperties() { + if (properties == null) { + properties = new LinkedHashMap<>(); + } + return properties; + } + + private Object getProperty(String name) { + return properties != null ? properties.get(name) : null; + } + + private static void checkPropertyName(String name) { + if (name == null || name.isEmpty()) { + throw new IllegalArgumentException("Property name must not be null or empty"); + } + } + + private MessageFormatRuntimeException invalidConversion(String name, Object value, String type) { + return new MessageFormatRuntimeException( + "Property " + name + " was " + value.getClass().getName() + " and cannot be read as " + type); + } +} diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledProducer.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledProducer.java index 1d1176f5257..cc652f86815 100644 --- a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledProducer.java +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledProducer.java @@ -36,6 +36,7 @@ public class PooledProducer implements MessageProducer { private boolean disableMessageTimestamp; private int priority; private long timeToLive; + private long deliveryDelay; private boolean anonymous = true; public PooledProducer(MessageProducer messageProducer, Destination destination) throws JMSException { @@ -112,26 +113,39 @@ public void send(Destination destination, Message message, int deliveryMode, int */ @Override public void send(Message message, CompletionListener completionListener) throws JMSException { - throw new UnsupportedOperationException("send(Message, CompletionListener) is not supported"); - + send(destination, message, getDeliveryMode(), getPriority(), getTimeToLive(), completionListener); } @Override public void send(Message message, int deliveryMode, int priority, long timeToLive, CompletionListener completionListener) throws JMSException { - throw new UnsupportedOperationException("send(Message, deliveryMode, priority, timetoLive, CompletionListener) is not supported"); + send(destination, message, deliveryMode, priority, timeToLive, completionListener); } @Override public void send(Destination destination, Message message, CompletionListener completionListener) throws JMSException { - throw new UnsupportedOperationException("send(Destination, Message, CompletionListener) is not supported"); + send(destination, message, getDeliveryMode(), getPriority(), getTimeToLive(), completionListener); } @Override public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive, CompletionListener completionListener) throws JMSException { - throw new UnsupportedOperationException("send(Destination, Message, deliveryMode, priority, timetoLive, CompletionListener) is not supported"); + if (destination == null) { + if (messageProducer.getDestination() == null) { + throw new UnsupportedOperationException("A destination must be specified."); + } + throw new InvalidDestinationException("Don't understand null destinations"); + } + + MessageProducer messageProducer = getMessageProducer(); + + synchronized (messageProducer) { + if (anonymous && this.destination != null && !this.destination.equals(destination)) { + throw new UnsupportedOperationException("This producer can only send messages to: " + this.destination); + } + messageProducer.send(destination, message, deliveryMode, priority, timeToLive, completionListener); + } } /** @@ -144,19 +158,12 @@ public void send(Destination destination, Message message, int deliveryMode, int */ @Override public void setDeliveryDelay(long deliveryDelay) throws JMSException { - throw new UnsupportedOperationException("setDeliveryDelay() is not supported"); + this.deliveryDelay = deliveryDelay; } - /** - * Gets the delivery delay value for this MessageProducer. - * - * @return the delivery delay for this messageProducer - * @throws jakarta.jms.JMSException if the JMS provider fails to determine if deliver delay is - * disabled due to some internal error. - */ @Override public long getDeliveryDelay() throws JMSException { - throw new UnsupportedOperationException("getDeliveryDelay() is not supported"); + return deliveryDelay; } @Override diff --git a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledSession.java b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledSession.java index 729d2aa3b3f..d0190a91b8c 100644 --- a/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledSession.java +++ b/activemq-jms-pool/src/main/java/org/apache/activemq/jms/pool/PooledSession.java @@ -97,19 +97,19 @@ public void close() throws JMSException { } if (closed.compareAndSet(false, true)) { - boolean invalidate = false; + var invalidate = false; try { // lets reset the session getInternalSession().setMessageListener(null); // Close any consumers and browsers that may have been created. - for (Iterator iter = consumers.iterator(); iter.hasNext();) { - MessageConsumer consumer = iter.next(); + for (var iter = consumers.iterator(); iter.hasNext();) { + var consumer = iter.next(); consumer.close(); } - for (Iterator iter = browsers.iterator(); iter.hasNext();) { - QueueBrowser browser = iter.next(); + for (var iter = browsers.iterator(); iter.hasNext();) { + var browser = iter.next(); browser.close(); } @@ -152,7 +152,7 @@ public void close() throws JMSException { try { sessionPool.returnObject(key, sessionHolder); } catch (Exception e) { - jakarta.jms.IllegalStateException illegalStateException = new jakarta.jms.IllegalStateException(e.toString()); + var illegalStateException = new jakarta.jms.IllegalStateException(e.toString()); illegalStateException.initCause(e); throw illegalStateException; } @@ -272,7 +272,7 @@ public void rollback() throws JMSException { @Override public XAResource getXAResource() { - SessionHolder session = safeGetSessionHolder(); + var session = safeGetSessionHolder(); if (session.getSession() instanceof XASession) { return ((XASession) session.getSession()).getXAResource(); @@ -288,7 +288,7 @@ public Session getSession() { @Override public void run() { - SessionHolder session = safeGetSessionHolder(); + var session = safeGetSessionHolder(); if (session != null) { session.getSession().run(); } @@ -363,32 +363,32 @@ public QueueReceiver createReceiver(Queue queue, String selector) throws JMSExce @Override public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName) throws JMSException { - throw new UnsupportedOperationException("createSharedConsumer(Topic, sharedSubscriptionName) is not supported"); + return addConsumer(getInternalSession().createSharedConsumer(topic, sharedSubscriptionName)); } @Override public MessageConsumer createSharedConsumer(Topic topic, String sharedSubscriptionName, String messageSelector) throws JMSException { - throw new UnsupportedOperationException("createSharedConsumer(Topic, sharedSubscriptionName, messageSelector) is not supported"); + return addConsumer(getInternalSession().createSharedConsumer(topic, sharedSubscriptionName, messageSelector)); } @Override public MessageConsumer createDurableConsumer(Topic topic, String name) throws JMSException { - throw new UnsupportedOperationException("createDurableConsumer(Topic, name) is not supported"); + return addConsumer(getInternalSession().createDurableConsumer(topic, name)); } @Override public MessageConsumer createDurableConsumer(Topic topic, String name, String messageSelector, boolean noLocal) throws JMSException { - throw new UnsupportedOperationException("createDurableConsumer(Topic, name, messageSelector, noLocal) is not supported"); + return addConsumer(getInternalSession().createDurableConsumer(topic, name, messageSelector, noLocal)); } @Override public MessageConsumer createSharedDurableConsumer(Topic topic, String name) throws JMSException { - throw new UnsupportedOperationException("createSharedDurableConsumer(Topic, name) is not supported"); + return addConsumer(getInternalSession().createSharedDurableConsumer(topic, name)); } @Override public MessageConsumer createSharedDurableConsumer(Topic topic, String name, String messageSelector) throws JMSException { - throw new UnsupportedOperationException("createSharedDurableConsumer(Topic, name, messageSelector) is not supported"); + return addConsumer(getInternalSession().createSharedDurableConsumer(topic, name, messageSelector)); } // Producer related methods @@ -487,6 +487,22 @@ public void setIsXa(boolean isXa) { this.isXa = isXa; } + /** + * @return true if this session hands out a single cached anonymous MessageProducer + * instead of creating a new MessageProducer for each request. + */ + public boolean isUseAnonymousProducers() { + return useAnonymousProducers; + } + + /** + * @return true if this pooled session facade has been closed and its internal + * session handed back to the pool. + */ + public boolean isClosed() { + return closed.get(); + } + @Override public String toString() { return "PooledSession { " + safeGetSessionHolder() + " }"; diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupportTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupportTest.java new file mode 100644 index 00000000000..0ba5f9e9d2b --- /dev/null +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/JmsPoolExceptionSupportTest.java @@ -0,0 +1,74 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertSame; + +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; + +import org.junit.Test; + +public class JmsPoolExceptionSupportTest { + + @Test(timeout = 60000) + public void testToRuntimeExceptionMapsAllSpecTypes() { + assertMapped(new jakarta.jms.IllegalStateException("m", "c"), IllegalStateRuntimeException.class); + assertMapped(new InvalidClientIDException("m", "c"), InvalidClientIDRuntimeException.class); + assertMapped(new InvalidDestinationException("m", "c"), InvalidDestinationRuntimeException.class); + assertMapped(new InvalidSelectorException("m", "c"), InvalidSelectorRuntimeException.class); + assertMapped(new JMSSecurityException("m", "c"), JMSSecurityRuntimeException.class); + assertMapped(new MessageFormatException("m", "c"), MessageFormatRuntimeException.class); + assertMapped(new MessageNotWriteableException("m", "c"), MessageNotWriteableRuntimeException.class); + assertMapped(new ResourceAllocationException("m", "c"), ResourceAllocationRuntimeException.class); + assertMapped(new TransactionInProgressException("m", "c"), TransactionInProgressRuntimeException.class); + assertMapped(new TransactionRolledBackException("m", "c"), TransactionRolledBackRuntimeException.class); + + // the base type and unrecognized subtypes map to the generic runtime exception + assertMapped(new JMSException("m", "c"), JMSRuntimeException.class); + assertMapped(new JMSException("m", "c") {}, JMSRuntimeException.class); + } + + private static void assertMapped(JMSException cause, Class expectedType) { + var mapped = JmsPoolExceptionSupport.toRuntimeException(cause); + assertEquals("Wrong runtime exception type for " + cause.getClass().getSimpleName(), + expectedType, mapped.getClass()); + assertEquals(cause.getMessage(), mapped.getMessage()); + assertEquals(cause.getErrorCode(), mapped.getErrorCode()); + assertSame(cause, mapped.getCause()); + } +} diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledConnectionSecurityExceptionTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledConnectionSecurityExceptionTest.java index 9927c365d9f..439cd681df6 100644 --- a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledConnectionSecurityExceptionTest.java +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledConnectionSecurityExceptionTest.java @@ -18,7 +18,10 @@ import jakarta.jms.Connection; import jakarta.jms.ExceptionListener; +import jakarta.jms.JMSConsumer; +import jakarta.jms.JMSContext; import jakarta.jms.JMSException; +import jakarta.jms.JMSRuntimeException; import jakarta.jms.JMSSecurityException; import jakarta.jms.MessageProducer; import jakarta.jms.Queue; @@ -46,6 +49,7 @@ import java.util.List; import java.util.concurrent.CountDownLatch; +import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThrows; @@ -226,6 +230,39 @@ public void testFailedCreateConsumerConnectionStillWorks() throws JMSException { } } + @Test + public void testCreateContextWithInvalidCredentials() throws Exception { + try (final JMSContext context = pooledConnFact.createContext("invalid", "credentials")) { + JMSRuntimeException thrown = null; + try { + context.createProducer(); + fail("Expected JMSRuntimeException for invalid credentials"); + } catch (JMSRuntimeException e) { + thrown = e; + } + + // ConnectionPool.createSession wraps the failed session create, so walk the + // cause chain to prove the failure is the broker's authentication rejection. + boolean securityCause = false; + for (Throwable t = thrown; t != null; t = t.getCause()) { + if (t instanceof JMSSecurityException) { + securityCause = true; + break; + } + } + assertTrue("Expected JMSSecurityException in the cause chain but was: " + thrown, securityCause); + } + + // the factory must still hand out working contexts for valid credentials + try (final JMSContext context = pooledConnFact.createContext("system", "manager")) { + final Queue queue = context.createQueue(name.getMethodName()); + context.createProducer().send(queue, "ok"); + try (final JMSConsumer consumer = context.createConsumer(queue)) { + assertEquals("ok", consumer.receiveBody(String.class, 5000)); + } + } + } + public String getName() { return name.getMethodName(); } diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSConsumerTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSConsumerTest.java new file mode 100644 index 00000000000..c3b5ad03f9c --- /dev/null +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSConsumerTest.java @@ -0,0 +1,269 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import jakarta.jms.IllegalStateRuntimeException; +import jakarta.jms.JMSException; +import jakarta.jms.Message; +import jakarta.jms.MessageListener; +import jakarta.jms.Session; +import jakarta.jms.TextMessage; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class PooledJMSConsumerTest extends JmsPoolTestSupport { + + private ActiveMQConnectionFactory factory; + private PooledConnectionFactory pooledFactory; + private String connectionUri; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.setUseJmx(false); + brokerService.setAdvisorySupport(false); + brokerService.setSchedulerSupport(false); + var connector = brokerService.addConnector("tcp://localhost:0"); + brokerService.start(); + + connectionUri = connector.getPublishableConnectString(); + factory = new ActiveMQConnectionFactory(connectionUri); + pooledFactory = new PooledConnectionFactory(); + pooledFactory.setConnectionFactory(factory); + pooledFactory.setMaxConnections(1); + } + + @Override + @After + public void tearDown() throws Exception { + try { + pooledFactory.stop(); + } catch (Exception ex) { + // ignored + } + super.tearDown(); + } + + @Test(timeout = 60000) + public void testReceive() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.receive"); + context.createProducer().send(queue, "hello"); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receive(5000); + assertNotNull("Should have received a message", msg); + } + } + } + + @Test(timeout = 60000) + public void testReceiveNoWaitReturnsNull() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.nowait"); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receiveNoWait(); + assertNull("Should not have received a message", msg); + } + } + } + + @Test(timeout = 60000) + public void testReceiveBody() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.body"); + context.createProducer().send(queue, "payload"); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("payload", body); + } + } + } + + @Test(timeout = 60000) + public void testReceiveBodyNoWaitReturnsNull() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.body.nowait"); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBodyNoWait(String.class); + assertNull(body); + } + } + } + + @Test(timeout = 60000) + public void testClose() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.close"); + var consumer = context.createConsumer(queue); + consumer.close(); + } + } + + @Test(timeout = 60000) + public void testMessageListenerDelivery() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.listener"); + var consumer = context.createConsumer(queue); + + final var delivered = new CountDownLatch(3); + final var bodies = Collections.synchronizedList(new ArrayList()); + MessageListener listener = message -> { + try { + bodies.add(((TextMessage) message).getText()); + } catch (JMSException e) { + // body stays missing and the containsAll assert below fails + } + delivered.countDown(); + }; + + consumer.setMessageListener(listener); + assertSame(listener, consumer.getMessageListener()); + + var producer = context.createProducer(); + producer.send(queue, "one"); + producer.send(queue, "two"); + producer.send(queue, "three"); + + assertTrue("Listener should have received all messages", + delivered.await(10, TimeUnit.SECONDS)); + assertTrue("Listener should have seen all bodies: " + bodies, + bodies.containsAll(Arrays.asList("one", "two", "three"))); + consumer.close(); + } + } + + @Test(timeout = 60000) + public void testMessageListenerDeliveryNotTrackedForAcknowledge() throws Exception { + // Listener-delivered messages are not tracked by the context's acknowledge() + // support (documented limitation), so without an explicit ack on the message + // itself the delivery stays unacknowledged and the broker redelivers it. + try (var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE)) { + var queue = context.createQueue("test.consumer.listener.ack"); + context.createProducer().send(queue, "needs explicit ack"); + + var consumer = context.createConsumer(queue); + final var delivered = new CountDownLatch(1); + consumer.setMessageListener(message -> delivered.countDown()); + assertTrue(delivered.await(10, TimeUnit.SECONDS)); + + // no-op for listener-delivered messages: the context never saw them + context.acknowledge(); + consumer.close(); + } + + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.listener.ack"); + try (var consumer = context.createConsumer(queue)) { + var redelivered = consumer.receive(5000); + assertNotNull("Unacknowledged listener-delivered message should be redelivered", redelivered); + assertTrue(redelivered.getJMSRedelivered()); + } + } + } + + @Test(timeout = 60000) + public void testMessageListenerExplicitMessageAcknowledge() throws Exception { + // The documented workaround: acknowledge the delivered Message directly + try (var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE)) { + var queue = context.createQueue("test.consumer.listener.msgack"); + context.createProducer().send(queue, "ack in listener"); + + var consumer = context.createConsumer(queue); + final var acked = new CountDownLatch(1); + consumer.setMessageListener(message -> { + try { + message.acknowledge(); + acked.countDown(); + } catch (JMSException e) { + // latch never counts down and the await below fails + } + }); + assertTrue("Listener should have acknowledged the message", + acked.await(10, TimeUnit.SECONDS)); + consumer.close(); + } + + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.listener.msgack"); + try (var consumer = context.createConsumer(queue)) { + assertNull("Acknowledged message must not be redelivered", consumer.receive(1000)); + } + } + } + + @Test(timeout = 60000) + public void testConsumerUseAfterCloseThrows() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.consumer.use.after.close"); + var consumer = context.createConsumer(queue); + consumer.close(); + + // close is idempotent + consumer.close(); + + try { + consumer.receive(100); + fail("Expected IllegalStateRuntimeException from receive on closed consumer"); + } catch (IllegalStateRuntimeException expected) { + } + + try { + consumer.receiveNoWait(); + fail("Expected IllegalStateRuntimeException from receiveNoWait on closed consumer"); + } catch (IllegalStateRuntimeException expected) { + } + + try { + consumer.receiveBodyNoWait(String.class); + fail("Expected IllegalStateRuntimeException from receiveBodyNoWait on closed consumer"); + } catch (IllegalStateRuntimeException expected) { + } + + try { + consumer.getMessageSelector(); + fail("Expected IllegalStateRuntimeException from getMessageSelector on closed consumer"); + } catch (IllegalStateRuntimeException expected) { + } + } + } + +} diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextBrokerFailureTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextBrokerFailureTest.java new file mode 100644 index 00000000000..a72131c2e79 --- /dev/null +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextBrokerFailureTest.java @@ -0,0 +1,295 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertTrue; + +import java.io.IOException; + +import jakarta.jms.Connection; +import jakarta.jms.IllegalStateRuntimeException; +import jakarta.jms.JMSContext; +import jakarta.jms.JMSRuntimeException; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.apache.activemq.transport.mock.MockTransport; +import org.apache.activemq.util.Wait; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +/** + * Exercises the JMS 2.0 pooled JMSContext surface under transport and broker + * failure: operations must fail with mapped runtime exceptions, and the pool's + * reconnectOnException support (default true) must evict the dead connection so + * the factory hands out working contexts again. + */ +public class PooledJMSContextBrokerFailureTest extends JmsPoolTestSupport { + + private ActiveMQConnectionFactory factory; + private PooledConnectionFactory pooledFactory; + private int brokerPort; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.setUseJmx(false); + brokerService.setAdvisorySupport(false); + brokerService.setSchedulerSupport(false); + var connector = brokerService.addConnector("tcp://localhost:0"); + brokerService.start(); + brokerService.waitUntilStarted(); + + brokerPort = connector.getConnectUri().getPort(); + + // MockTransport in the chain lets tests inject transport failures deterministically + factory = new ActiveMQConnectionFactory("mock:" + connector.getConnectUri() + "?closeAsync=false"); + pooledFactory = new PooledConnectionFactory(); + pooledFactory.setConnectionFactory(factory); + pooledFactory.setMaxConnections(1); + } + + @Override + @After + public void tearDown() throws Exception { + try { + pooledFactory.stop(); + } catch (Exception ex) { + // ignored + } + super.tearDown(); + } + + @Test(timeout = 60000) + public void testSendFailsAfterTransportFailure() throws Exception { + var context = pooledFactory.createContext(); + try { + var queue = context.createQueue("test.context.failure.send"); + context.createProducer().send(queue, "before failure"); + + injectTransportFailure(); + + // Failure propagation is async. Transitional failures may surface as the + // generic JMSRuntimeException (ConnectionFailedException); once the pool's + // ExceptionListener has closed the connection the settled state is the + // mapped IllegalStateRuntimeException. + assertTrue("Send should settle into IllegalStateRuntimeException after transport failure", + Wait.waitFor(() -> { + try { + context.createProducer().send(queue, "should fail"); + return false; + } catch (IllegalStateRuntimeException settled) { + return true; + } catch (JMSRuntimeException transitional) { + return false; + } + }, 5000, 10)); + } finally { + closeQuietly(context); + } + } + + @Test(timeout = 60000) + public void testReceiveFailsAfterTransportFailure() throws Exception { + var context = pooledFactory.createContext(); + try { + var queue = context.createQueue("test.context.failure.receive"); + context.createProducer().send(queue, "delivered"); + + var consumer = context.createConsumer(queue); + assertEquals("delivered", consumer.receiveBody(String.class, 5000)); + + injectTransportFailure(); + + // Until the async teardown closes the consumer, receive can simply return + // null (no broker interaction with an empty prefetch); afterwards it must + // throw the mapped IllegalStateRuntimeException. + assertTrue("Receive should settle into IllegalStateRuntimeException after transport failure", + Wait.waitFor(() -> { + try { + consumer.receive(50); + return false; + } catch (IllegalStateRuntimeException settled) { + return true; + } catch (JMSRuntimeException transitional) { + return false; + } + }, 5000, 10)); + } finally { + closeQuietly(context); + } + } + + @Test(timeout = 60000) + public void testPoolEvictsFailedConnectionAndRecovers() throws Exception { + var before = probeUnderlyingConnection(); + + var context = pooledFactory.createContext(); + var queue = context.createQueue("test.context.failure.recovery"); + context.createProducer().send(queue, "before failure"); + + injectTransportFailure(); + + assertTrue("Context should start failing after transport failure", + Wait.waitFor(() -> { + try { + context.createProducer().send(queue, "should fail"); + return false; + } catch (JMSRuntimeException expected) { + return true; + } + }, 5000, 10)); + + closeQuietly(context); + + // reconnectOnException (default true) must evict the dead connection and + // let the factory produce a fully working context again + assertTrue("Pool should provide a working context after eviction", + Wait.waitFor(() -> { + try (var fresh = pooledFactory.createContext()) { + var freshQueue = fresh.createQueue("test.context.failure.recovery.fresh"); + fresh.createProducer().send(freshQueue, "recovered"); + try (var consumer = fresh.createConsumer(freshQueue)) { + return "recovered".equals(consumer.receiveBody(String.class, 2000)); + } + } catch (Exception e) { + return false; + } + }, 10000, 10)); + + var after = probeUnderlyingConnection(); + assertNotSame("Failed connection should have been evicted from the pool", before, after); + } + + @Test(timeout = 60000) + public void testContextCloseAfterTransportFailureReleasesPool() throws Exception { + var context = pooledFactory.createContext(); + var queue = context.createQueue("test.context.failure.close"); + context.createProducer().send(queue, "before failure"); + + injectTransportFailure(); + + assertTrue("Context should start failing after transport failure", + Wait.waitFor(() -> { + try { + context.createProducer().send(queue, "should fail"); + return false; + } catch (JMSRuntimeException expected) { + return true; + } + }, 5000, 10)); + + // close may or may not throw depending on how far the async teardown got; + // either way the pool references must be released + closeQuietly(context); + + assertTrue("A fresh context must work after closing the failed one", + Wait.waitFor(() -> { + try (var fresh = pooledFactory.createContext()) { + fresh.createProducer().send(fresh.createQueue("test.context.failure.close.fresh"), "ok"); + return true; + } catch (Exception e) { + return false; + } + }, 10000, 10)); + } + + @Test(timeout = 60000) + public void testBrokerRestartNewContextRecovers() throws Exception { + var context = pooledFactory.createContext(); + var queue = context.createQueue("test.context.failure.restart"); + context.createProducer().send(queue, "before restart"); + + brokerService.stop(); + brokerService.waitUntilStopped(); + + assertTrue("Context should fail after broker stop", + Wait.waitFor(() -> { + try { + context.createProducer().send(queue, "should fail"); + return false; + } catch (JMSRuntimeException expected) { + return true; + } + }, 10000, 10)); + + closeQuietly(context); + + // restart on the same port so the pooled factory's URI stays valid; reassign + // the inherited field so JmsPoolTestSupport.tearDown stops the new instance + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.setUseJmx(false); + brokerService.setAdvisorySupport(false); + brokerService.setSchedulerSupport(false); + brokerService.addConnector("tcp://localhost:" + brokerPort); + brokerService.start(); + brokerService.waitUntilStarted(); + + assertTrue("A fresh context must recover after broker restart", + Wait.waitFor(() -> { + try (var fresh = pooledFactory.createContext()) { + var freshQueue = fresh.createQueue("test.context.failure.restart.fresh"); + fresh.createProducer().send(freshQueue, "recovered"); + try (var consumer = fresh.createConsumer(freshQueue)) { + return "recovered".equals(consumer.receiveBody(String.class, 2000)); + } + } catch (Exception e) { + return false; + } + }, 15000, 10)); + } + + /** + * Fires an IOException into the MockTransport of the pool's current underlying + * connection, simulating a transport drop. + */ + private void injectTransportFailure() throws Exception { + var amqConnection = (ActiveMQConnection) probeUnderlyingConnection(); + var transport = (MockTransport) amqConnection.getTransportChannel().narrow(MockTransport.class); + transport.onException(new IOException("simulated transport failure")); + } + + /** + * Borrows a pooled connection facade (sharing the same keyed ConnectionPool the + * contexts use, maxConnections=1) to observe the current underlying connection. + */ + private Connection probeUnderlyingConnection() throws Exception { + var probe = (PooledConnection) pooledFactory.createConnection(); + try { + return probe.getConnection(); + } finally { + probe.close(); + } + } + + private void closeQuietly(JMSContext context) { + try { + context.close(); + } catch (JMSRuntimeException ignored) { + // closing a context whose transport already failed may legitimately throw + } + } +} diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextTest.java new file mode 100644 index 00000000000..a6aed1d128a --- /dev/null +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSContextTest.java @@ -0,0 +1,647 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNotSame; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.lang.reflect.InvocationTargetException; +import java.lang.reflect.Proxy; +import java.util.concurrent.CyclicBarrier; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; + +import jakarta.jms.Connection; +import jakarta.jms.IllegalStateRuntimeException; +import jakarta.jms.JMSException; +import jakarta.jms.JMSRuntimeException; +import jakarta.jms.Message; +import jakarta.jms.MessageFormatRuntimeException; +import jakarta.jms.Session; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class PooledJMSContextTest extends JmsPoolTestSupport { + + private ActiveMQConnectionFactory factory; + private PooledConnectionFactory pooledFactory; + private String connectionUri; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.setUseJmx(false); + brokerService.setAdvisorySupport(false); + brokerService.setSchedulerSupport(false); + var connector = brokerService.addConnector("tcp://localhost:0"); + brokerService.start(); + + connectionUri = connector.getPublishableConnectString(); + factory = new ActiveMQConnectionFactory(connectionUri); + pooledFactory = new PooledConnectionFactory(); + pooledFactory.setConnectionFactory(factory); + pooledFactory.setMaxConnections(1); + } + + @Override + @After + public void tearDown() throws Exception { + try { + pooledFactory.stop(); + } catch (Exception ex) { + // ignored + } + super.tearDown(); + } + + @Test(timeout = 60000) + public void testCreateContext() throws Exception { + var context = pooledFactory.createContext(); + assertNotNull(context); + context.close(); + } + + @Test(timeout = 60000) + public void testCreateContextWithSessionMode() throws Exception { + var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE); + assertNotNull(context); + assertEquals(Session.CLIENT_ACKNOWLEDGE, context.getSessionMode()); + assertFalse(context.getTransacted()); + context.close(); + } + + @Test(timeout = 60000) + public void testCreateContextTransacted() throws Exception { + var context = pooledFactory.createContext(Session.SESSION_TRANSACTED); + assertNotNull(context); + assertEquals(Session.SESSION_TRANSACTED, context.getSessionMode()); + assertTrue(context.getTransacted()); + context.close(); + } + + @Test(timeout = 60000) + public void testCreateContextWithCredentials() throws Exception { + var context = pooledFactory.createContext(null, null); + assertNotNull(context); + context.close(); + } + + @Test(timeout = 60000) + public void testCreateContextWithCredentialsAndSessionMode() throws Exception { + var context = pooledFactory.createContext(null, null, Session.AUTO_ACKNOWLEDGE); + assertNotNull(context); + context.close(); + } + + @Test(timeout = 60000) + public void testAutoStart() throws Exception { + try (var context = pooledFactory.createContext()) { + assertTrue(context.getAutoStart()); + context.setAutoStart(false); + assertFalse(context.getAutoStart()); + } + } + + @Test(timeout = 60000) + public void testStartStop() throws Exception { + try (var context = pooledFactory.createContext()) { + context.start(); + context.stop(); + } + } + + @Test(timeout = 60000) + public void testGetMetaData() throws Exception { + try (var context = pooledFactory.createContext()) { + assertNotNull(context.getMetaData()); + } + } + + @Test(timeout = 60000) + public void testCreateMessageTypes() throws Exception { + try (var context = pooledFactory.createContext()) { + var msg = context.createMessage(); + assertNotNull(msg); + + var textMsg = context.createTextMessage(); + assertNotNull(textMsg); + + var textMsgWithBody = context.createTextMessage("hello"); + assertNotNull(textMsgWithBody); + assertEquals("hello", textMsgWithBody.getText()); + + var bytesMsg = context.createBytesMessage(); + assertNotNull(bytesMsg); + + var mapMsg = context.createMapMessage(); + assertNotNull(mapMsg); + + var objMsg = context.createObjectMessage(); + assertNotNull(objMsg); + + var streamMsg = context.createStreamMessage(); + assertNotNull(streamMsg); + } + } + + @Test(timeout = 60000) + public void testCreateQueue() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.queue"); + assertNotNull(queue); + assertEquals("test.context.queue", queue.getQueueName()); + } + } + + @Test(timeout = 60000) + public void testCreateTopic() throws Exception { + try (var context = pooledFactory.createContext()) { + var topic = context.createTopic("test.context.topic"); + assertNotNull(topic); + assertEquals("test.context.topic", topic.getTopicName()); + } + } + + @Test(timeout = 60000) + public void testCreateTemporaryQueue() throws Exception { + try (var context = pooledFactory.createContext()) { + var tempQueue = context.createTemporaryQueue(); + assertNotNull(tempQueue); + } + } + + @Test(timeout = 60000) + public void testCreateTemporaryTopic() throws Exception { + try (var context = pooledFactory.createContext()) { + var tempTopic = context.createTemporaryTopic(); + assertNotNull(tempTopic); + } + } + + @Test(timeout = 60000) + public void testCreateProducer() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer(); + assertNotNull(producer); + } + } + + @Test(timeout = 60000) + public void testCreateConsumer() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.consumer"); + var consumer = context.createConsumer(queue); + assertNotNull(consumer); + consumer.close(); + } + } + + @Test(timeout = 60000) + public void testCreateConsumerWithSelector() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.consumer.sel"); + var consumer = context.createConsumer(queue, "color = 'red'"); + assertNotNull(consumer); + consumer.close(); + } + } + + @Test(timeout = 60000) + public void testCreateConsumerWithSelectorAndNoLocal() throws Exception { + try (var context = pooledFactory.createContext()) { + var topic = context.createTopic("test.context.consumer.nolocal"); + var consumer = context.createConsumer(topic, null, true); + assertNotNull(consumer); + consumer.close(); + } + } + + @Test(timeout = 60000) + public void testSendAndReceive() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.sendrecv"); + context.createProducer().send(queue, "message body"); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("message body", body); + } + } + } + + @Test(timeout = 60000) + public void testTransactedCommit() throws Exception { + try (var context = pooledFactory.createContext(Session.SESSION_TRANSACTED)) { + var queue = context.createQueue("test.context.txcommit"); + context.createProducer().send(queue, "tx msg"); + context.commit(); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("tx msg", body); + } + } + } + + @Test(timeout = 60000) + public void testTransactedRollback() throws Exception { + try (var txContext = pooledFactory.createContext(Session.SESSION_TRANSACTED)) { + var queue = txContext.createQueue("test.context.txrollback"); + txContext.createProducer().send(queue, "will be rolled back"); + txContext.rollback(); + } + + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.txrollback"); + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receiveNoWait(); + assertTrue("Message should have been rolled back", msg == null); + } + } + } + + @Test(timeout = 60000) + public void testChildContext() throws Exception { + var parent = pooledFactory.createContext(); + var child = parent.createContext(Session.AUTO_ACKNOWLEDGE); + assertNotNull(child); + + var queue = child.createQueue("test.context.child"); + child.createProducer().send(queue, "from child"); + + try (var consumer = child.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("from child", body); + } + + child.close(); + parent.close(); + } + + @Test(timeout = 60000) + public void testDoubleCloseIsHarmless() throws Exception { + var context = pooledFactory.createContext(); + context.close(); + context.close(); + } + + @Test(timeout = 60000, expected = IllegalStateRuntimeException.class) + public void testUseAfterCloseThrows() throws Exception { + var context = pooledFactory.createContext(); + context.close(); + context.createQueue("should.fail"); + } + + @Test(timeout = 60000) + public void testCreateDurableConsumer() throws Exception { + try (var context = pooledFactory.createContext()) { + context.setClientID("durable-test-client"); + var topic = context.createTopic("test.context.durable"); + var consumer = context.createDurableConsumer(topic, "durSub"); + assertNotNull(consumer); + consumer.close(); + context.unsubscribe("durSub"); + } + } + + @Test(timeout = 60000) + public void testCreateBrowser() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.browser"); + context.createProducer().send(queue, "browse me"); + + var browser = context.createBrowser(queue); + assertNotNull(browser); + browser.close(); + } + } + + @Test(timeout = 60000) + public void testCreateBrowserWithSelector() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.browser.sel"); + var browser = context.createBrowser(queue, "type = 'A'"); + assertNotNull(browser); + browser.close(); + } + } + + @Test(timeout = 60000) + public void testExceptionListener() throws Exception { + try (var context = pooledFactory.createContext()) { + context.setExceptionListener(e -> LOG.warn("Exception: {}", e.getMessage())); + assertNotNull(context.getExceptionListener()); + } + } + + @Test(timeout = 60000) + public void testSessionExhaustionThroughContexts() throws Exception { + pooledFactory.setMaximumActiveSessionPerConnection(1); + pooledFactory.setBlockIfSessionPoolIsFull(false); + + var context1 = pooledFactory.createContext(); + var context2 = pooledFactory.createContext(); + try { + var queue = context1.createQueue("test.context.exhaustion"); + context1.createProducer().send(queue, "one"); + + // context1 holds the only pooled session, so context2 cannot get one + try { + context2.createProducer(); + fail("Expected session pool exhaustion to surface as IllegalStateRuntimeException"); + } catch (IllegalStateRuntimeException expected) { + } + + // closing context1 returns its session to the pool and unblocks context2 + context1.close(); + context2.createProducer().send(queue, "two"); + + try (var consumer = context2.createConsumer(queue)) { + assertEquals("one", consumer.receiveBody(String.class, 5000)); + assertEquals("two", consumer.receiveBody(String.class, 5000)); + } + } finally { + context1.close(); + context2.close(); + } + } + + @Test(timeout = 60000) + public void testExpiredConnectionEvictsFromPoolWithContexts() throws Exception { + pooledFactory.setExpiryTimeout(10); + + var before = probeUnderlyingConnection(); + + var context = pooledFactory.createContext(); + var queue = context.createQueue("test.context.expiry"); + context.createProducer().send(queue, "before expiry"); + + // let the underlying connection expire while the context still holds it + TimeUnit.MILLISECONDS.sleep(500); + context.close(); + + // the next context must trigger eviction and get a fresh underlying connection + try (var context2 = pooledFactory.createContext()) { + context2.createProducer().send(context2.createQueue("test.context.expiry"), "after expiry"); + var after = probeUnderlyingConnection(); + assertNotSame("expired connection should have been evicted from the pool", before, after); + } + } + + /** + * Borrows a pooled connection facade from the factory (sharing the same + * keyed ConnectionPool the contexts use) to observe the identity of the + * current underlying provider connection. + */ + private Connection probeUnderlyingConnection() throws Exception { + var probe = (PooledConnection) pooledFactory.createConnection(); + try { + return probe.getConnection(); + } finally { + probe.close(); + } + } + + @Test(timeout = 60000) + public void testAcknowledgeAcksConsumedMessages() throws Exception { + try (var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE)) { + var queue = context.createQueue("test.context.acknowledge"); + context.createProducer().send(queue, "ack me"); + + try (var consumer = context.createConsumer(queue)) { + var received = consumer.receive(5000); + assertNotNull(received); + context.acknowledge(); + } + } + + // the acknowledged message must not be redelivered to a new consumer + try (var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE)) { + var queue = context.createQueue("test.context.acknowledge"); + try (var consumer = context.createConsumer(queue)) { + assertNull("Acknowledged message must not be redelivered", consumer.receive(1000)); + } + } + } + + @Test(timeout = 60000) + public void testAcknowledgeWithoutReceiveIsNoOp() throws Exception { + try (var context = pooledFactory.createContext(Session.CLIENT_ACKNOWLEDGE)) { + context.acknowledge(); + } + } + + @Test(timeout = 60000) + public void testCommitOnNonTransactedThrowsIllegalStateRuntimeException() throws Exception { + try (var context = pooledFactory.createContext()) { + try { + context.commit(); + fail("Expected IllegalStateRuntimeException"); + } catch (IllegalStateRuntimeException expected) { + } + } + } + + @Test(timeout = 60000) + public void testCreateContextAfterFactoryStopThrows() throws Exception { + var stopped = new PooledConnectionFactory(); + stopped.setConnectionFactory(factory); + stopped.stop(); + try { + stopped.createContext(); + fail("Expected IllegalStateRuntimeException"); + } catch (IllegalStateRuntimeException expected) { + } + } + + @Test(timeout = 60000) + public void testStartAfterStop() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.startafterstop"); + context.stop(); + context.start(); + + context.createProducer().send(queue, "after restart"); + try (var consumer = context.createConsumer(queue)) { + assertEquals("after restart", consumer.receiveBody(String.class, 5000)); + } + } + } + + @Test(timeout = 60000) + public void testReceiveBodyWrongTypeThrowsMessageFormatRuntimeException() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.context.receivebody.wrongtype"); + context.createProducer().send(queue, "not a long"); + try (var consumer = context.createConsumer(queue)) { + try { + consumer.receiveBody(Long.class, 5000); + fail("Expected MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException expected) { + } + } + } + } + + @Test(timeout = 60000) + public void testCreateSharedConsumerUnsupportedByClient() throws Exception { + try (var context = pooledFactory.createContext()) { + var topic = context.createTopic("test.context.shared"); + try { + context.createSharedConsumer(topic, "sharedSub"); + fail("Expected UnsupportedOperationException from ActiveMQ Classic client"); + } catch (UnsupportedOperationException expected) { + } + } + } + + @Test(timeout = 60000) + public void testConcurrentChildCloseDoesNotReleaseParentConnection() throws Exception { + try (var parent = pooledFactory.createContext()) { + for (int i = 0; i < 25; i++) { + final var child = parent.createContext(Session.AUTO_ACKNOWLEDGE); + final var barrier = new CyclicBarrier(2); + Runnable closer = () -> { + try { + barrier.await(5, TimeUnit.SECONDS); + } catch (Exception ignored) { + } + child.close(); + }; + var t1 = new Thread(closer); + var t2 = new Thread(closer); + t1.start(); + t2.start(); + t1.join(10000); + t2.join(10000); + + assertNotNull("Parent context must stay usable after concurrent child close (iteration " + i + ")", + parent.createTextMessage()); + } + } + } + + @Test(timeout = 60000) + public void testCreateProducerWithoutAnonymousProducers() throws Exception { + var nonAnon = new PooledConnectionFactory(); + nonAnon.setConnectionFactory(factory); + nonAnon.setMaxConnections(1); + nonAnon.setUseAnonymousProducers(false); + try { + try (var context = nonAnon.createContext()) { + var queue = context.createQueue("test.context.nonanon"); + for (int i = 0; i < 5; i++) { + context.createProducer().send(queue, "msg" + i); + } + try (var consumer = context.createConsumer(queue)) { + for (int i = 0; i < 5; i++) { + assertNotNull(consumer.receive(5000)); + } + } + } + } finally { + nonAnon.stop(); + } + } + + @Test(timeout = 60000) + public void testCreateProducerDoesNotStartConnection() throws Exception { + var pool = new ConnectionPool(factory.createConnection()); + pool.incrementReferenceCount(); + var connection = new PooledConnection(pool); + + var context = new PooledJMSContext(connection, Session.AUTO_ACKNOWLEDGE); + try { + context.createProducer(); + assertFalse("Creating a producer must not auto-start the connection", + ((ActiveMQConnection) connection.getConnection()).isStarted()); + + var queue = context.createQueue("test.context.autostart"); + context.createConsumer(queue).close(); + assertTrue("Creating a consumer must auto-start the connection", + ((ActiveMQConnection) connection.getConnection()).isStarted()); + } finally { + context.close(); + } + } + + @Test(timeout = 60000) + public void testConnectionReleasedWhenSessionCloseFails() throws Exception { + var pool = new ConnectionPool(factory.createConnection()); + pool.incrementReferenceCount(); + var connection = new SessionCloseFailsConnection(pool); + + var context = new PooledJMSContext(connection, Session.AUTO_ACKNOWLEDGE); + assertNotNull(context.createTextMessage()); // force lazy session creation + + try { + context.close(); + fail("Expected JMSRuntimeException from failing session close"); + } catch (JMSRuntimeException expected) { + } + + assertTrue("Connection must be released even when session close fails", + connection.connectionClosed.get()); + } + + private static class SessionCloseFailsConnection extends PooledConnection { + + final AtomicBoolean connectionClosed = new AtomicBoolean(); + + SessionCloseFailsConnection(ConnectionPool pool) { + super(pool); + } + + @Override + public Session createSession(boolean transacted, int ackMode) throws JMSException { + final var real = super.createSession(transacted, ackMode); + return (Session) Proxy.newProxyInstance( + getClass().getClassLoader(), + new Class[] { Session.class }, + (proxy, method, args) -> { + if ("close".equals(method.getName())) { + throw new JMSException("Simulated session close failure"); + } + try { + return method.invoke(real, args); + } catch (InvocationTargetException e) { + throw e.getCause(); + } + }); + } + + @Override + public void close() throws JMSException { + connectionClosed.set(true); + super.close(); + } + } +} diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSProducerTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSProducerTest.java new file mode 100644 index 00000000000..f93c9234e5b --- /dev/null +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/PooledJMSProducerTest.java @@ -0,0 +1,438 @@ +/** + * 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 static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertSame; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.util.HashMap; +import java.util.Map; + +import jakarta.jms.CompletionListener; +import jakarta.jms.DeliveryMode; +import jakarta.jms.JMSRuntimeException; +import jakarta.jms.Message; +import jakarta.jms.MessageFormatRuntimeException; + +import org.apache.activemq.ActiveMQConnectionFactory; +import org.apache.activemq.broker.BrokerService; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +public class PooledJMSProducerTest extends JmsPoolTestSupport { + + private ActiveMQConnectionFactory factory; + private PooledConnectionFactory pooledFactory; + private String connectionUri; + + @Override + @Before + public void setUp() throws Exception { + super.setUp(); + + brokerService = new BrokerService(); + brokerService.setPersistent(false); + brokerService.setUseJmx(false); + brokerService.setAdvisorySupport(false); + brokerService.setSchedulerSupport(false); + var connector = brokerService.addConnector("tcp://localhost:0"); + brokerService.start(); + + connectionUri = connector.getPublishableConnectString(); + factory = new ActiveMQConnectionFactory(connectionUri); + pooledFactory = new PooledConnectionFactory(); + pooledFactory.setConnectionFactory(factory); + pooledFactory.setMaxConnections(1); + } + + @Override + @After + public void tearDown() throws Exception { + try { + pooledFactory.stop(); + } catch (Exception ex) { + // ignored + } + super.tearDown(); + } + + @Test(timeout = 60000) + public void testSendTextMessage() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.text"); + context.createProducer().send(queue, "hello world"); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("hello world", body); + } + } + } + + @Test(timeout = 60000) + public void testSendMapMessage() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.map"); + // Deliberately not 'var': HashMap is both Map and Serializable, and the + // declared Map type selects the send(Destination, Map) overload. + Map body = new HashMap<>(); + body.put("key1", "value1"); + body.put("key2", 42); + context.createProducer().send(queue, body); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receive(5000); + assertNotNull(msg); + assertTrue(msg instanceof jakarta.jms.MapMessage); + assertEquals("value1", ((jakarta.jms.MapMessage) msg).getString("key1")); + assertEquals(42, ((jakarta.jms.MapMessage) msg).getInt("key2")); + } + } + } + + @Test(timeout = 60000) + public void testSendBytesMessage() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.bytes"); + var payload = new byte[]{1, 2, 3, 4, 5}; + context.createProducer().send(queue, payload); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receive(5000); + assertNotNull(msg); + assertTrue(msg instanceof jakarta.jms.BytesMessage); + } + } + } + + @Test(timeout = 60000) + public void testSendMessageObject() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.message"); + var msg = context.createTextMessage("direct message"); + context.createProducer().send(queue, msg); + + try (var consumer = context.createConsumer(queue)) { + var body = consumer.receiveBody(String.class, 5000); + assertEquals("direct message", body); + } + } + } + + @Test(timeout = 60000) + public void testFluentQoSSetters() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer(); + + var returned = producer + .setDeliveryMode(DeliveryMode.NON_PERSISTENT) + .setPriority(7) + .setTimeToLive(30000) + .setDeliveryDelay(1000) + .setDisableMessageID(true) + .setDisableMessageTimestamp(true); + + assertEquals(producer, returned); + assertEquals(DeliveryMode.NON_PERSISTENT, producer.getDeliveryMode()); + assertEquals(7, producer.getPriority()); + assertEquals(30000, producer.getTimeToLive()); + assertEquals(1000, producer.getDeliveryDelay()); + assertTrue(producer.getDisableMessageID()); + assertTrue(producer.getDisableMessageTimestamp()); + } + } + + @Test(timeout = 60000) + public void testDefaultQoSValues() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer(); + + assertEquals(DeliveryMode.PERSISTENT, producer.getDeliveryMode()); + assertEquals(4, producer.getPriority()); + assertEquals(0, producer.getTimeToLive()); + assertEquals(0, producer.getDeliveryDelay()); + assertFalse(producer.getDisableMessageID()); + assertFalse(producer.getDisableMessageTimestamp()); + assertNull(producer.getAsync()); + } + } + + @Test(timeout = 60000, expected = JMSRuntimeException.class) + public void testSetInvalidDeliveryMode() throws Exception { + try (var context = pooledFactory.createContext()) { + context.createProducer().setDeliveryMode(99); + } + } + + @Test(timeout = 60000, expected = JMSRuntimeException.class) + public void testSetInvalidPriority() throws Exception { + try (var context = pooledFactory.createContext()) { + context.createProducer().setPriority(10); + } + } + + @Test(timeout = 60000, expected = MessageFormatRuntimeException.class) + public void testSendNullMessage() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.null"); + context.createProducer().send(queue, (Message) null); + } + } + + @Test(timeout = 60000) + public void testJMSHeaders() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.headers"); + var replyTo = context.createQueue("test.producer.replyto"); + + context.createProducer() + .setJMSCorrelationID("corr-123") + .setJMSType("myType") + .setJMSReplyTo(replyTo) + .send(queue, "with headers"); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receive(5000); + assertNotNull(msg); + assertEquals("corr-123", msg.getJMSCorrelationID()); + assertEquals("myType", msg.getJMSType()); + assertNotNull(msg.getJMSReplyTo()); + } + } + } + + @Test(timeout = 60000) + public void testMessageProperties() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.props"); + + var producer = context.createProducer() + .setProperty("strProp", "hello") + .setProperty("intProp", 42) + .setProperty("boolProp", true) + .setProperty("longProp", 100L) + .setProperty("doubleProp", 3.14); + + assertTrue(producer.propertyExists("strProp")); + assertFalse(producer.propertyExists("nonexistent")); + assertEquals("hello", producer.getStringProperty("strProp")); + assertEquals(42, producer.getIntProperty("intProp")); + assertTrue(producer.getBooleanProperty("boolProp")); + assertEquals(100L, producer.getLongProperty("longProp")); + assertEquals(3.14, producer.getDoubleProperty("doubleProp"), 0.001); + + producer.send(queue, "with props"); + + try (var consumer = context.createConsumer(queue)) { + var msg = consumer.receive(5000); + assertNotNull(msg); + assertEquals("hello", msg.getStringProperty("strProp")); + assertEquals(42, msg.getIntProperty("intProp")); + assertTrue(msg.getBooleanProperty("boolProp")); + } + } + } + + @Test(timeout = 60000) + public void testClearProperties() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer() + .setProperty("key", "value"); + assertTrue(producer.propertyExists("key")); + + producer.clearProperties(); + assertFalse(producer.propertyExists("key")); + } + } + + @Test(timeout = 60000) + public void testGetPropertyNames() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer() + .setProperty("a", 1) + .setProperty("b", 2); + var names = producer.getPropertyNames(); + assertEquals(2, names.size()); + assertTrue(names.contains("a")); + assertTrue(names.contains("b")); + } + } + + @Test(timeout = 60000) + public void testGetPropertyNamesEmpty() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer(); + var names = producer.getPropertyNames(); + assertTrue(names.isEmpty()); + } + } + + @Test(timeout = 60000) + public void testCorrelationIDAsBytes() throws Exception { + try (var context = pooledFactory.createContext()) { + var corrId = new byte[]{10, 20, 30}; + var producer = context.createProducer() + .setJMSCorrelationIDAsBytes(corrId); + var result = producer.getJMSCorrelationIDAsBytes(); + assertEquals(3, result.length); + assertEquals(10, result[0]); + } + } + + @Test(timeout = 60000) + public void testPropertyStringConversion() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer() + .setProperty("num", "42"); + assertEquals(42, producer.getIntProperty("num")); + assertEquals(42L, producer.getLongProperty("num")); + } + } + + @Test(timeout = 60000) + public void testNumericPropertyWidening() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer() + .setProperty("byteProp", (byte) 5) + .setProperty("shortProp", (short) 6) + .setProperty("intProp", 7) + .setProperty("floatProp", 1.5f); + + assertEquals(5, producer.getShortProperty("byteProp")); + assertEquals(5, producer.getIntProperty("byteProp")); + assertEquals(5L, producer.getLongProperty("byteProp")); + assertEquals(6, producer.getIntProperty("shortProp")); + assertEquals(6L, producer.getLongProperty("shortProp")); + assertEquals(7L, producer.getLongProperty("intProp")); + assertEquals(1.5d, producer.getDoubleProperty("floatProp"), 0.0); + } + } + + @Test(timeout = 60000) + public void testInvalidPropertyConversions() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer().setProperty("intProp", 7); + + try { + producer.getByteProperty("intProp"); // narrowing is not a valid conversion + fail("Expected MessageFormatRuntimeException"); + } catch (MessageFormatRuntimeException expected) { + } + + try { + producer.getIntProperty("missing"); // null numeric read + fail("Expected NumberFormatException"); + } catch (NumberFormatException expected) { + } + + try { + producer.getFloatProperty("missing"); // null float read + fail("Expected NullPointerException"); + } catch (NullPointerException expected) { + } + + assertFalse(producer.getBooleanProperty("missing")); + assertNull(producer.getStringProperty("missing")); + assertEquals("7", producer.getStringProperty("intProp")); + } + } + + @Test(timeout = 60000) + public void testProducerUseAfterContextCloseThrows() throws Exception { + var context = pooledFactory.createContext(); + var queue = context.createQueue("test.producer.use.after.close"); + var producer = context.createProducer(); + var prebuilt = context.createTextMessage("prebuilt"); + producer.send(queue, "before close"); + context.close(); + + // locally stored producer state remains readable after the context closes + assertEquals(DeliveryMode.PERSISTENT, producer.getDeliveryMode()); + + try { + producer.send(queue, "after close"); + fail("Expected IllegalStateRuntimeException sending through a closed context's producer"); + } catch (jakarta.jms.IllegalStateRuntimeException expected) { + } + + // the pre-built message path must not silently send on the pooled session + try { + producer.send(queue, prebuilt); + fail("Expected IllegalStateRuntimeException sending a pre-built message through a closed context's producer"); + } catch (jakarta.jms.IllegalStateRuntimeException expected) { + } + } + + @Test(timeout = 60000) + public void testAsyncSendPassesListenerToProvider() throws Exception { + try (var context = pooledFactory.createContext()) { + var queue = context.createQueue("test.producer.async"); + var producer = context.createProducer(); + + var listener = new CompletionListener() { + @Override + public void onCompletion(Message message) { + } + + @Override + public void onException(Message message, Exception exception) { + } + }; + + assertSame(producer, producer.setAsync(listener)); + assertSame(listener, producer.getAsync()); + + try { + producer.send(queue, "async message"); + fail("Expected UnsupportedOperationException from ActiveMQ Classic client"); + } catch (UnsupportedOperationException expected) { + } + + // clearing the listener restores synchronous sends + producer.setAsync(null); + producer.send(queue, "sync message"); + try (var consumer = context.createConsumer(queue)) { + assertEquals("sync message", consumer.receiveBody(String.class, 5000)); + } + } + } + + @Test(timeout = 60000) + public void testSetPropertyNameValidation() throws Exception { + try (var context = pooledFactory.createContext()) { + var producer = context.createProducer(); + + try { + producer.setProperty(null, "value"); + fail("Expected IllegalArgumentException for null property name"); + } catch (IllegalArgumentException expected) { + } + + try { + producer.setProperty("", 1); + fail("Expected IllegalArgumentException for empty property name"); + } catch (IllegalArgumentException expected) { + } + } + } +} diff --git a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/XAConnectionPoolTest.java b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/XAConnectionPoolTest.java index 6655ec014d8..8d48b5260cd 100644 --- a/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/XAConnectionPoolTest.java +++ b/activemq-jms-pool/src/test/java/org/apache/activemq/jms/pool/XAConnectionPoolTest.java @@ -32,6 +32,7 @@ import jakarta.jms.QueueConnectionFactory; import jakarta.jms.QueueSender; import jakarta.jms.QueueSession; +import jakarta.jms.JMSContext; import jakarta.jms.Session; import jakarta.jms.TopicConnection; import jakarta.jms.TopicConnectionFactory; @@ -174,6 +175,233 @@ public Transaction suspend() throws SystemException { pcf.stop(); } + @Test(timeout = 60000) + public void testCreateContextSpansTransactions() throws Exception { + final Vector syncs = new Vector(); + final java.util.concurrent.atomic.AtomicReference currentXid = + new java.util.concurrent.atomic.AtomicReference(createXid()); + final java.util.concurrent.atomic.AtomicReference enlisted = + new java.util.concurrent.atomic.AtomicReference(); + XaPooledConnectionFactory pcf = new XaPooledConnectionFactory(); + pcf.setConnectionFactory(new XAConnectionFactoryOnly(new ActiveMQXAConnectionFactory("vm://test?broker.persistent=false"))); + + // simple TM that is always in a tx, enlists with the current xid and tracks syncs + pcf.setTransactionManager(new TransactionManager() { + @Override + public void begin() throws NotSupportedException, SystemException { + } + + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, IllegalStateException, RollbackException, SecurityException, SystemException { + } + + @Override + public int getStatus() throws SystemException { + return Status.STATUS_ACTIVE; + } + + @Override + public Transaction getTransaction() throws SystemException { + return new Transaction() { + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, RollbackException, SecurityException, SystemException { + } + + @Override + public boolean delistResource(XAResource xaRes, int flag) throws IllegalStateException, SystemException { + return false; + } + + @Override + public boolean enlistResource(XAResource xaRes) throws IllegalStateException, RollbackException, SystemException { + try { + xaRes.start(currentXid.get(), 0); + enlisted.set(xaRes); + } catch (XAException e) { + throw new SystemException(e.getMessage()); + } + return true; + } + + @Override + public int getStatus() throws SystemException { + return 0; + } + + @Override + public void registerSynchronization(Synchronization synch) throws IllegalStateException, RollbackException, SystemException { + syncs.add(synch); + } + + @Override + public void rollback() throws IllegalStateException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + }; + } + + @Override + public void resume(Transaction tobj) throws IllegalStateException, InvalidTransactionException, SystemException { + } + + @Override + public void rollback() throws IllegalStateException, SecurityException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + + @Override + public void setTransactionTimeout(int seconds) throws SystemException { + } + + @Override + public Transaction suspend() throws SystemException { + return null; + } + }); + + JMSContext context = pcf.createContext(); + try { + jakarta.jms.Queue queue = context.createQueue("test.xa.context.two.tx"); + + // first transaction + Xid firstXid = currentXid.get(); + context.createProducer().send(queue, "one"); + assertEquals("First transaction should have enlisted one session", 1, syncs.size()); + enlisted.get().end(firstXid, XAResource.TMSUCCESS); + enlisted.get().rollback(firstXid); + syncs.get(0).beforeCompletion(); + syncs.get(0).afterCompletion(1); + + // second transaction - the context must renew its pooled session and enlist again + Xid secondXid = createXid(); + currentXid.set(secondXid); + context.createProducer().send(queue, "two"); + assertEquals("Second transaction should have enlisted a renewed session", 2, syncs.size()); + enlisted.get().end(secondXid, XAResource.TMSUCCESS); + enlisted.get().rollback(secondXid); + syncs.get(1).beforeCompletion(); + syncs.get(1).afterCompletion(1); + } finally { + context.close(); + pcf.stop(); + } + } + + @Test(timeout = 60000) + public void testCreateContextEnlistsSessionInTransaction() throws Exception { + final Vector syncs = new Vector(); + XaPooledConnectionFactory pcf = new XaPooledConnectionFactory(); + pcf.setConnectionFactory(new XAConnectionFactoryOnly(new ActiveMQXAConnectionFactory("vm://test?broker.persistent=false"))); + + final Xid xid = createXid(); + + // simple TM that is in a tx, enlists resources and tracks syncs + pcf.setTransactionManager(new TransactionManager() { + @Override + public void begin() throws NotSupportedException, SystemException { + } + + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, IllegalStateException, RollbackException, SecurityException, SystemException { + } + + @Override + public int getStatus() throws SystemException { + return Status.STATUS_ACTIVE; + } + + @Override + public Transaction getTransaction() throws SystemException { + return new Transaction() { + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, RollbackException, SecurityException, SystemException { + } + + @Override + public boolean delistResource(XAResource xaRes, int flag) throws IllegalStateException, SystemException { + return false; + } + + @Override + public boolean enlistResource(XAResource xaRes) throws IllegalStateException, RollbackException, SystemException { + try { + xaRes.start(xid, 0); + } catch (XAException e) { + throw new SystemException(e.getMessage()); + } + return true; + } + + @Override + public int getStatus() throws SystemException { + return 0; + } + + @Override + public void registerSynchronization(Synchronization synch) throws IllegalStateException, RollbackException, SystemException { + syncs.add(synch); + } + + @Override + public void rollback() throws IllegalStateException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + }; + } + + @Override + public void resume(Transaction tobj) throws IllegalStateException, InvalidTransactionException, SystemException { + } + + @Override + public void rollback() throws IllegalStateException, SecurityException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + + @Override + public void setTransactionTimeout(int seconds) throws SystemException { + } + + @Override + public Transaction suspend() throws SystemException { + return null; + } + }); + + JMSContext context = pcf.createContext(); + try { + jakarta.jms.Queue queue = context.createQueue("test.xa.context"); + context.createProducer().send(queue, "xa message"); + + // the send succeeding proves the XA session was enlisted; the + // registered synchronization proves it is tied to the transaction + assertFalse("Session should have been enlisted in the transaction", syncs.isEmpty()); + + // simulate a commit + for (Synchronization sync : syncs) { + sync.beforeCompletion(); + } + for (Synchronization sync : syncs) { + sync.afterCompletion(1); + } + } finally { + context.close(); + pcf.stop(); + } + } + static long txGenerator = 22; public Xid createXid() throws IOException { diff --git a/activemq-pool/src/test/java/org/apache/activemq/pool/XAConnectionPoolTest.java b/activemq-pool/src/test/java/org/apache/activemq/pool/XAConnectionPoolTest.java index 9dcd4b0fd14..3cc47105770 100644 --- a/activemq-pool/src/test/java/org/apache/activemq/pool/XAConnectionPoolTest.java +++ b/activemq-pool/src/test/java/org/apache/activemq/pool/XAConnectionPoolTest.java @@ -21,7 +21,10 @@ import java.io.IOException; import java.util.Hashtable; import java.util.Vector; +import java.util.concurrent.atomic.AtomicReference; +import jakarta.jms.JMSContext; +import jakarta.jms.Queue; import jakarta.jms.QueueConnection; import jakarta.jms.QueueConnectionFactory; import jakarta.jms.QueueSender; @@ -397,4 +400,161 @@ public Transaction suspend() throws SystemException { connection.close(); pcf.stop(); } + + private TransactionManager trackingTransactionManager(final Vector syncs, final AtomicReference currentXid, + final AtomicReference enlisted) { + return new TransactionManager() { + @Override + public void begin() throws NotSupportedException, SystemException { + } + + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, IllegalStateException, RollbackException, SecurityException, SystemException { + } + + @Override + public int getStatus() throws SystemException { + return Status.STATUS_ACTIVE; + } + + @Override + public Transaction getTransaction() throws SystemException { + return new Transaction() { + @Override + public void commit() throws HeuristicMixedException, HeuristicRollbackException, RollbackException, SecurityException, SystemException { + } + + @Override + public boolean delistResource(XAResource xaRes, int flag) throws IllegalStateException, SystemException { + return false; + } + + @Override + public boolean enlistResource(XAResource xaRes) throws IllegalStateException, RollbackException, SystemException { + try { + xaRes.start(currentXid.get(), 0); + enlisted.set(xaRes); + } catch (XAException e) { + SystemException se = new SystemException("XAException errorCode=" + e.errorCode); + se.initCause(e); + throw se; + } + return true; + } + + @Override + public int getStatus() throws SystemException { + return 0; + } + + @Override + public void registerSynchronization(Synchronization synch) throws IllegalStateException, RollbackException, SystemException { + syncs.add(synch); + } + + @Override + public void rollback() throws IllegalStateException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + }; + } + + @Override + public void resume(Transaction tobj) throws IllegalStateException, InvalidTransactionException, SystemException { + } + + @Override + public void rollback() throws IllegalStateException, SecurityException, SystemException { + } + + @Override + public void setRollbackOnly() throws IllegalStateException, SystemException { + } + + @Override + public void setTransactionTimeout(int seconds) throws SystemException { + } + + @Override + public Transaction suspend() throws SystemException { + return null; + } + }; + } + + public void testCreateContextSpansTransactions() throws Exception { + final Vector syncs = new Vector(); + final AtomicReference currentXid = new AtomicReference(createXid()); + final AtomicReference enlisted = new AtomicReference(); + XaPooledConnectionFactory pcf = new XaPooledConnectionFactory(); + pcf.setConnectionFactory(new ActiveMQXAConnectionFactory("vm://xaConnectionPoolTest?broker.persistent=false")); + pcf.setTransactionManager(trackingTransactionManager(syncs, currentXid, enlisted)); + + JMSContext context = pcf.createContext(); + try { + Queue queue = context.createQueue("test.xa.pool.context"); + + // first transaction + Xid firstXid = currentXid.get(); + context.createProducer().send(queue, "one"); + assertEquals("first transaction should have enlisted one session", 1, syncs.size()); + enlisted.get().end(firstXid, XAResource.TMSUCCESS); + enlisted.get().rollback(firstXid); + syncs.get(0).beforeCompletion(); + syncs.get(0).afterCompletion(1); + + // second transaction - the context must renew its pooled session and enlist again + Xid secondXid = createXid(); + currentXid.set(secondXid); + context.createProducer().send(queue, "two"); + assertEquals("second transaction should have enlisted a renewed session", 2, syncs.size()); + enlisted.get().end(secondXid, XAResource.TMSUCCESS); + enlisted.get().rollback(secondXid); + syncs.get(1).beforeCompletion(); + syncs.get(1).afterCompletion(1); + } finally { + context.close(); + pcf.stop(); + } + } + + public void testJcaCreateContextSpansTransactions() throws Exception { + final Vector syncs = new Vector(); + final AtomicReference currentXid = new AtomicReference(createXid()); + final AtomicReference enlisted = new AtomicReference(); + JcaPooledConnectionFactory pcf = new JcaPooledConnectionFactory(); + pcf.setName("jca-pool-context-test"); + pcf.setConnectionFactory(new ActiveMQXAConnectionFactory("vm://xaConnectionPoolTest?broker.persistent=false")); + pcf.setTransactionManager(trackingTransactionManager(syncs, currentXid, enlisted)); + + JMSContext context = pcf.createContext(); + try { + Queue queue = context.createQueue("test.jca.pool.context"); + + // first transaction + Xid firstXid = currentXid.get(); + context.createProducer().send(queue, "one"); + assertEquals("first transaction should have enlisted one session", 1, syncs.size()); + enlisted.get().end(firstXid, XAResource.TMSUCCESS); + enlisted.get().rollback(firstXid); + syncs.get(0).beforeCompletion(); + syncs.get(0).afterCompletion(1); + + // second transaction - the context must renew its pooled session and enlist again + Xid secondXid = createXid(); + currentXid.set(secondXid); + context.createProducer().send(queue, "two"); + assertEquals("second transaction should have enlisted a renewed session", 2, syncs.size()); + enlisted.get().end(secondXid, XAResource.TMSUCCESS); + enlisted.get().rollback(secondXid); + syncs.get(1).beforeCompletion(); + syncs.get(1).afterCompletion(1); + } finally { + context.close(); + pcf.stop(); + } + } }