<commit_before>/*
 * 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.SystemException;
import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import javax.transaction.xa.XAResource;
import java.io.PrintWriter;
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.jboss.logging.Logger;

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

   private static final Logger logger = Logger.getLogger(ActiveMQRAManagedConnection.class);

   /**
    * 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 TransactionManager tm;

   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()) {
         ActiveMQRALogger.LOGGER.trace("constructor(" + mcf + ", " + cri + ", " + userName + ", ****)");
      }

      this.mcf = mcf;
      this.cri = cri;
      this.tm = ra.getTM();
      this.ra = ra;
      this.userName = userName;
      this.password = password;
      eventListeners = Collections.synchronizedList(new ArrayList<ConnectionEventListener>());
      handles = Collections.synchronizedSet(new HashSet<ActiveMQRASession>());

      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 {
      if (logger.isTraceEnabled()) {
         ActiveMQRALogger.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 {
      if (logger.isTraceEnabled()) {
         ActiveMQRALogger.LOGGER.trace("destroyHandles()");
      }

      try {
         if (connection != null) {
            connection.stop();
         }
      } catch (Throwable t) {
         logger.trace("Ignored error stopping connection", t);
      }

      for (ActiveMQRASession session : handles) {<commit_msg>ARTEMIS-1305 Avoid deadlock when shutting down RA and reconnects<commit_after>/*
 * 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.SystemException;
import javax.transaction.Transaction;
import javax.transaction.TransactionManager;
import javax.transaction.xa.XAResource;
import java.io.PrintWriter;
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.jboss.logging.Logger;

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

   private static final Logger logger = Logger.getLogger(ActiveMQRAManagedConnection.class);

   /**
    * 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 TransactionManager tm;

   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()) {
         ActiveMQRALogger.LOGGER.trace("constructor(" + mcf + ", " + cri + ", " + userName + ", ****)");
      }

      this.mcf = mcf;
      this.cri = cri;
      this.tm = ra.getTM();
      this.ra = ra;
      this.userName = userName;
      this.password = password;
      eventListeners = Collections.synchronizedList(new ArrayList<ConnectionEventListener>());
      handles = Collections.synchronizedSet(new HashSet<ActiveMQRASession>());

      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 {
      if (logger.isTraceEnabled()) {
         ActiveMQRALogger.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 {
      if (logger.isTraceEnabled()) {
         ActiveMQRALogger.LOGGER.trace("destroyHandles()");
      }

      for (ActiveMQRASession session : handles) {
{'fraction_non_alphanumeric': 0.08452587936338103, 'fraction_numerical': 0.0008024608800320985, 'mean_word_length': 3.068280739934712, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 12, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6444413', 'n_tokens_mistral': 3964, 'n_tokens_neox': 3841, 'n_words': 1111}