ActiveMQRAManagedConnection.java

/*
 * 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.artemis.ra;

import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.ResourceAllocationException;
import javax.jms.Session;
import javax.jms.XASession;
import javax.resource.ResourceException;
import javax.resource.spi.ConnectionEvent;
import javax.resource.spi.ConnectionEventListener;
import javax.resource.spi.ConnectionRequestInfo;
import javax.resource.spi.IllegalStateException;
import javax.resource.spi.LocalTransaction;
import javax.resource.spi.ManagedConnection;
import javax.resource.spi.ManagedConnectionMetaData;
import javax.resource.spi.SecurityException;
import javax.security.auth.Subject;
import javax.transaction.Status;
import javax.transaction.TransactionSynchronizationRegistry;
import javax.transaction.xa.XAResource;
import java.io.PrintWriter;
import java.lang.invoke.MethodHandles;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.ReentrantLock;

import org.apache.activemq.artemis.core.client.impl.ClientSessionInternal;
import org.apache.activemq.artemis.jms.client.ActiveMQConnection;
import org.apache.activemq.artemis.jms.client.ActiveMQConnectionFactory;
import org.apache.activemq.artemis.jms.client.ActiveMQXAConnection;
import org.apache.activemq.artemis.service.extensions.ServiceUtils;
import org.apache.activemq.artemis.service.extensions.xa.ActiveMQXAResourceWrapper;
import org.apache.activemq.artemis.utils.VersionLoader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The managed connection
 */
public final class ActiveMQRAManagedConnection implements ManagedConnection, ExceptionListener {

   private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

   /**
    * The managed connection factory
    */
   private final ActiveMQRAManagedConnectionFactory mcf;

   /**
    * The connection request information
    */
   private final ActiveMQRAConnectionRequestInfo cri;

   /**
    * The resource adapter
    */
   private final ActiveMQResourceAdapter ra;

   /**
    * The user name
    */
   private final String userName;

   /**
    * The password
    */
   private final String password;

   /**
    * Has the connection been destroyed
    */
   private final AtomicBoolean isDestroyed = new AtomicBoolean(false);

   /**
    * Event listeners
    */
   private final List<ConnectionEventListener> eventListeners;

   /**
    * Handles
    */
   private final Set<ActiveMQRASession> handles;

   /**
    * Lock
    */
   private ReentrantLock lock = new ReentrantLock();

   // Physical connection stuff
   private ActiveMQConnectionFactory connectionFactory;

   private ActiveMQXAConnection connection;

   // The ManagedConnection will play with a XA and a NonXASession to couple with
   // cases where a commit is called on a non-XAed (or non-enlisted) case.
   private Session nonXAsession;

   private XASession xaSession;

   private XAResource xaResource;

   private final TransactionSynchronizationRegistry tsr;

   private boolean inManagedTx;

   /**
    * Constructor
    *
    * @param mcf      The managed connection factory
    * @param cri      The connection request information
    * @param userName The user name
    * @param password The password
    */
   public ActiveMQRAManagedConnection(final ActiveMQRAManagedConnectionFactory mcf,
                                      final ActiveMQRAConnectionRequestInfo cri,
                                      final ActiveMQResourceAdapter ra,
                                      final String userName,
                                      final String password) throws ResourceException {
      if (logger.isTraceEnabled()) {
         logger.trace("constructor({}, {}, {}, ****))", mcf, cri, userName);
      }

      this.mcf = mcf;
      this.cri = cri;
      this.tsr = ra.getTSR();
      this.ra = ra;
      this.userName = userName;
      this.password = password;
      eventListeners = Collections.synchronizedList(new ArrayList<>());
      handles = Collections.synchronizedSet(new HashSet<>());

      connection = null;
      nonXAsession = null;
      xaSession = null;
      xaResource = null;

      try {
         setup();
      } catch (ResourceException e) {
         try {
            destroy();
         } catch (Throwable ignored) {
         }

         throw e;
      } catch (Throwable t) {
         try {
            destroy();
         } catch (Throwable ignored) {
         }
         throw new ResourceException("Error during setup", t);
      }
   }

   /**
    * Get a connection
    *
    * @param subject       The security subject
    * @param cxRequestInfo The request info
    * @return The connection
    * @throws ResourceException Thrown if an error occurs
    */
   @Override
   public synchronized Object getConnection(final Subject subject,
                                            final ConnectionRequestInfo cxRequestInfo) throws ResourceException {
      logger.trace("getConnection({}, {})", subject, cxRequestInfo);

      // Check user first
      ActiveMQRACredential credential = ActiveMQRACredential.getCredential(mcf, subject, cxRequestInfo);

      // Null users are allowed!
      if (userName != null && !userName.equals(credential.getUserName())) {
         throw new SecurityException("Password credentials not the same, reauthentication not allowed");
      }

      if (userName == null && credential.getUserName() != null) {
         throw new SecurityException("Password credentials not the same, reauthentication not allowed");
      }

      if (isDestroyed.get()) {
         throw new IllegalStateException("The managed connection is already destroyed");
      }

      ActiveMQRASession session = new ActiveMQRASession(this, (ActiveMQRAConnectionRequestInfo) cxRequestInfo);
      handles.add(session);
      return session;
   }

   /**
    * Destroy all handles.
    *
    * @throws ResourceException Failed to close one or more handles.
    */
   private void destroyHandles() throws ResourceException {
      logger.trace("destroyHandles()");

      for (ActiveMQRASession session : handles) {
         session.destroy();
      }

      handles.clear();
   }

   /**
    * Destroy the physical connection.
    *
    * @throws ResourceException Could not property close the session and connection.
    */
   @Override
   public void destroy() throws ResourceException {
      logger.trace("destroy()");

      if (isDestroyed.get() || connection == null) {
         return;
      }

      isDestroyed.set(true);

      try {
         connection.setExceptionListener(null);
      } catch (JMSException e) {
         logger.debug("Error unsetting the exception listener {}", this, e);
      }

      connection.signalStopToAllSessions();

      try {
         // we must close the ActiveMQConnectionFactory because it contains a ServerLocator
         if (connectionFactory != null) {
            ra.closeConnectionFactory(mcf.getProperties());
         }
      } catch (Exception e) {
         logger.debug(e.getMessage(), e);
      }

      destroyHandles();

      try {
         // The following calls should not be necessary, as the connection should close the
         // ClientSessionFactory, which will close the sessions.
         try {
            /**
             * (xa|nonXA)Session.close() may NOT be called BEFORE connection.close()
             * <p>
             * If the ClientSessionFactory is trying to fail-over or reconnect with -1 attempts, and
             * one calls session.close() it may effectively dead-lock.
             * <p>
             * connection close will close the ClientSessionFactory which will close all sessions.
             */
            connection.close();

            if (nonXAsession != null) {
               nonXAsession.close();
            }

            if (xaSession != null) {
               xaSession.close();
            }
         } catch (JMSException e) {
            logger.debug("Error closing session {}", this, e);
         }

      } catch (Throwable e) {
         throw new ResourceException("Could not properly close the session and connection", e);
      }
   }

   /**
    * Cleanup
    *
    * @throws ResourceException Thrown if an error occurs
    */
   @Override
   public void cleanup() throws ResourceException {
      logger.trace("cleanup()");

      if (isDestroyed.get()) {
         throw new IllegalStateException("ManagedConnection already destroyed");
      }

      destroyHandles();

      inManagedTx = false;

      inManagedTx = false;

      // I'm recreating the lock object when we return to the pool
      // because it looks too nasty to expect the connection handle
      // to unlock properly in certain race conditions
      // where the dissociation of the managed connection is "random".
      lock = new ReentrantLock();
   }

   /**
    * Move a handler from one mc to this one.
    *
    * @param obj An object of type ActiveMQSession.
    * @throws ResourceException     Failed to associate connection.
    * @throws IllegalStateException ManagedConnection in an illegal state.
    */
   @Override
   public void associateConnection(final Object obj) throws ResourceException {
      logger.trace("associateConnection({})", obj);

      if (!isDestroyed.get() && obj instanceof ActiveMQRASession) {
         ActiveMQRASession h = (ActiveMQRASession) obj;
         h.setManagedConnection(this);
         handles.add(h);
      } else {
         throw new IllegalStateException("ManagedConnection in an illegal state");
      }
   }

   public void checkTransactionActive() throws JMSException {
      // don't bother looking at the transaction if there's an active XID
      if (!inManagedTx && tsr != null) {
         int status = tsr.getTransactionStatus();
         // Only allow states that will actually succeed
         if (status == Status.STATUS_COMMITTED || status == Status.STATUS_MARKED_ROLLBACK || status == Status.STATUS_ROLLEDBACK || status == Status.STATUS_ROLLING_BACK) {
            String statusMessage = "";
            if (status == Status.STATUS_COMMITTED) {
               statusMessage = "committed";
            } else if (status == Status.STATUS_MARKED_ROLLBACK) {
               statusMessage = "marked for rollback";
            } else if (status == Status.STATUS_ROLLEDBACK) {
               statusMessage = "rolled back";
            } else if (status == Status.STATUS_ROLLING_BACK) {
               statusMessage = "rolling back";
            }
            throw new javax.jms.IllegalStateException("Transaction is " + statusMessage);
         }
      }
   }

   /**
    * Aqquire a lock on the managed connection
    */
   protected void lock() {
      logger.trace("lock()");

      lock.lock();
   }

   /**
    * Aqquire a lock on the managed connection within the specified period
    *
    * @throws JMSException Thrown if an error occurs
    */
   protected void tryLock() throws JMSException {
      logger.trace("tryLock()");

      Integer tryLock = mcf.getUseTryLock();
      if (tryLock == null || tryLock.intValue() <= 0) {
         lock();
         return;
      }
      try {
         if (lock.tryLock(tryLock.intValue(), TimeUnit.SECONDS) == false) {
            throw new ResourceAllocationException("Unable to obtain lock in " + tryLock + " seconds: " + this);
         }
      } catch (InterruptedException e) {
         throw new ResourceAllocationException("Interrupted attempting lock: " + this);
      }
   }

   /**
    * Unlock the managed connection
    */
   protected void unlock() {
      logger.trace("unlock()");

      lock.unlock();
   }

   /**
    * Add a connection event listener.
    *
    * @param l The connection event listener to be added.
    */
   @Override
   public void addConnectionEventListener(final ConnectionEventListener l) {
      logger.trace("addConnectionEventListener({})", l);

      eventListeners.add(l);
   }

   /**
    * Remove a connection event listener.
    *
    * @param l The connection event listener to be removed.
    */
   @Override
   public void removeConnectionEventListener(final ConnectionEventListener l) {
      logger.trace("removeConnectionEventListener({})", l);

      eventListeners.remove(l);
   }

   /**
    * Get the XAResource for the connection.
    *
    * @return The XAResource for the connection.
    * @throws ResourceException XA transaction not supported
    */
   @Override
   public XAResource getXAResource() throws ResourceException {
      logger.trace("getXAResource()");

      //
      // Spec says a mc must always return the same XA resource,
      // so we cache it.
      //
      if (xaResource == null) {
         ClientSessionInternal csi = (ClientSessionInternal) xaSession.getXAResource();
         ActiveMQRAXAResource activeMQRAXAResource = new ActiveMQRAXAResource(this, xaSession.getXAResource());
         Map<String, Object> xaResourceProperties = new HashMap<>();
         xaResourceProperties.put(ActiveMQXAResourceWrapper.ACTIVEMQ_JNDI_NAME, ra.getJndiName());
         xaResourceProperties.put(ActiveMQXAResourceWrapper.ACTIVEMQ_NODE_ID, csi.getNodeId());
         xaResourceProperties.put(ActiveMQXAResourceWrapper.ACTIVEMQ_PRODUCT_NAME, ActiveMQResourceAdapter.PRODUCT_NAME);
         xaResourceProperties.put(ActiveMQXAResourceWrapper.ACTIVEMQ_PRODUCT_VERSION, VersionLoader.getVersion().getFullVersion());
         xaResource = ServiceUtils.wrapXAResource(activeMQRAXAResource, xaResourceProperties);
      }

      logger.trace("XAResource={}", xaResource);

      return xaResource;
   }

   /**
    * Get the location transaction for the connection.
    *
    * @return The local transaction for the connection.
    * @throws ResourceException Thrown if operation fails.
    */
   @Override
   public LocalTransaction getLocalTransaction() throws ResourceException {
      logger.trace("getLocalTransaction()");

      LocalTransaction tx = new ActiveMQRALocalTransaction(this);

      logger.trace("LocalTransaction={}", tx);

      return tx;
   }

   /**
    * Get the meta data for the connection.
    *
    * @return The meta data for the connection.
    * @throws ResourceException     Thrown if the operation fails.
    * @throws IllegalStateException Thrown if the managed connection already is destroyed.
    */
   @Override
   public ManagedConnectionMetaData getMetaData() throws ResourceException {
      logger.trace("getMetaData()");

      if (isDestroyed.get()) {
         throw new IllegalStateException("The managed connection is already destroyed");
      }

      return new ActiveMQRAMetaData(this);
   }

   /**
    * Set the log writer -- NOT SUPPORTED
    *
    * @param out The log writer
    * @throws ResourceException If operation fails
    */
   @Override
   public void setLogWriter(final PrintWriter out) throws ResourceException {
      logger.trace("setLogWriter({})", out);
   }

   /**
    * Get the log writer -- NOT SUPPORTED
    *
    * @return Always null
    * @throws ResourceException If operation fails
    */
   @Override
   public PrintWriter getLogWriter() throws ResourceException {
      logger.trace("getLogWriter()");

      return null;
   }

   /**
    * Notifies user of a JMS exception.
    *
    * @param exception The JMS exception
    */
   @Override
   public void onException(final JMSException exception) {
      if (ActiveMQConnection.EXCEPTION_FAILOVER.equals(exception.getErrorCode())) {
         return;
      }
      if (logger.isTraceEnabled()) {
         logger.trace("onException()", exception);
      }

      if (isDestroyed.get()) {
         if (logger.isTraceEnabled()) {
            logger.trace("Ignoring error on already destroyed connection {}", this, exception);
         }
         return;
      }

      ActiveMQRALogger.LOGGER.handlingJMSFailure(exception);

      try {
         connection.setExceptionListener(null);
      } catch (JMSException e) {
         logger.debug("Unable to unset exception listener", e);
      }

      ConnectionEvent event = new ConnectionEvent(this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, exception);
      sendEvent(event);
   }

   /**
    * Get the session for this connection.
    *
    * @return The session
    * @throws JMSException
    */
   protected Session getSession() throws JMSException {
      if (xaResource != null && inManagedTx) {
         logger.trace("getSession() -> XA session {}", xaSession.getSession());

         return xaSession.getSession();
      } else {
         logger.trace("getSession() -> non XA session {}", nonXAsession);

         return nonXAsession;
      }
   }

   /**
    * Send an event.
    *
    * @param event The event to send.
    */
   protected void sendEvent(final ConnectionEvent event) {
      logger.trace("sendEvent({})", event);

      int type = event.getId();

      // convert to an array to avoid concurrent modification exceptions
      ConnectionEventListener[] list = eventListeners.toArray(new ConnectionEventListener[eventListeners.size()]);

      for (ConnectionEventListener l : list) {
         switch (type) {
            case ConnectionEvent.CONNECTION_CLOSED:
               l.connectionClosed(event);
               break;

            case ConnectionEvent.LOCAL_TRANSACTION_STARTED:
               l.localTransactionStarted(event);
               break;

            case ConnectionEvent.LOCAL_TRANSACTION_COMMITTED:
               l.localTransactionCommitted(event);
               break;

            case ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK:
               l.localTransactionRolledback(event);
               break;

            case ConnectionEvent.CONNECTION_ERROR_OCCURRED:
               l.connectionErrorOccurred(event);
               break;

            default:
               throw new IllegalArgumentException("Illegal eventType: " + type);
         }
      }
   }

   /**
    * Remove a handle from the handle map.
    *
    * @param handle The handle to remove.
    */
   protected void removeHandle(final ActiveMQRASession handle) {
      logger.trace("removeHandle({})", handle);

      handles.remove(handle);
   }

   /**
    * Get the request info for this connection.
    *
    * @return The connection request info for this connection.
    */
   protected ActiveMQRAConnectionRequestInfo getCRI() {
      logger.trace("getCRI()");

      return cri;
   }

   /**
    * Get the connection factory for this connection.
    *
    * @return The connection factory for this connection.
    */
   protected ActiveMQRAManagedConnectionFactory getManagedConnectionFactory() {
      logger.trace("getManagedConnectionFactory()");

      return mcf;
   }

   /**
    * Start the connection
    *
    * @throws JMSException Thrown if the connection can't be started
    */
   void start() throws JMSException {
      logger.trace("start()");

      if (connection != null) {
         connection.start();
      }
   }

   /**
    * Stop the connection
    *
    * @throws JMSException Thrown if the connection can't be stopped
    */
   void stop() throws JMSException {
      logger.trace("stop()");

      if (connection != null) {
         connection.stop();
      }
   }

   /**
    * Get the user name
    *
    * @return The user name
    */
   protected String getUserName() {
      logger.trace("getUserName()");

      return userName;
   }

   /**
    * Setup the connection.
    *
    * @throws ResourceException Thrown if a connection couldn't be created
    */
   private void setup() throws ResourceException {
      logger.trace("setup()");

      try {

         createCF();

         boolean transacted = cri.isTransacted();
         int acknowledgeMode = Session.AUTO_ACKNOWLEDGE;
         if (cri.getType() == ActiveMQRAConnectionFactory.TOPIC_CONNECTION) {
            if (userName != null && password != null) {
               connection = (ActiveMQXAConnection) connectionFactory.createXATopicConnection(userName, password);
            } else {
               connection = (ActiveMQXAConnection) connectionFactory.createXATopicConnection();
            }

            connection.setExceptionListener(this);

            xaSession = connection.createXATopicSession();
            nonXAsession = connection.createNonXATopicSession(transacted, acknowledgeMode);

         } else if (cri.getType() == ActiveMQRAConnectionFactory.QUEUE_CONNECTION) {
            if (userName != null && password != null) {
               connection = (ActiveMQXAConnection) connectionFactory.createXAQueueConnection(userName, password);
            } else {
               connection = (ActiveMQXAConnection) connectionFactory.createXAQueueConnection();
            }

            connection.setExceptionListener(this);

            xaSession = connection.createXAQueueSession();
            nonXAsession = connection.createNonXAQueueSession(transacted, acknowledgeMode);

         } else {
            if (userName != null && password != null) {
               connection = (ActiveMQXAConnection) connectionFactory.createXAConnection(userName, password);
            } else {
               connection = (ActiveMQXAConnection) connectionFactory.createXAConnection();
            }

            connection.setExceptionListener(this);

            xaSession = connection.createXASession();
            nonXAsession = connection.createNonXASession(transacted, acknowledgeMode);
         }

      } catch (JMSException je) {
         throw new ResourceException(je.getMessage(), je);
      }
   }

   private void createCF() {
      if (connectionFactory == null) {
         connectionFactory = ra.getConnectionFactory(mcf.getProperties());
      }
   }

   protected void setInManagedTx(boolean inManagedTx) {
      this.inManagedTx = inManagedTx;
   }

   public ActiveMQConnectionFactory getConnectionFactory() {
      return connectionFactory;
   }
}