content: /**
   * 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.ra;
  
  import java.io.PrintWriter;
  import java.util.List;
  import java.util.concurrent.CopyOnWriteArrayList;
  import jakarta.jms.Connection;
  import jakarta.jms.ExceptionListener;
  import jakarta.jms.JMSException;
  import jakarta.resource.ResourceException;
  import jakarta.resource.spi.ConnectionEvent;
  import jakarta.resource.spi.ConnectionEventListener;
  import jakarta.resource.spi.ConnectionRequestInfo;
  import jakarta.resource.spi.LocalTransaction;
  import jakarta.resource.spi.ManagedConnection;
  import jakarta.resource.spi.ManagedConnectionMetaData;
  import javax.security.auth.Subject;
  import javax.transaction.xa.XAResource;
  import org.apache.activemq.ActiveMQConnection;
  import org.apache.activemq.LocalTransactionEventListener;
  import org.apache.activemq.TransactionContext;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * ActiveMQManagedConnection maps to real physical connection to the server.
   * Since a ManagedConnection has to provide a transaction managment interface to
   * the physical connection, and sessions are the objects implement transaction
   * managment interfaces in the JMS API, this object also maps to a singe
   * physical JMS session. <p/> The side-effect is that JMS connection the
   * application gets will allways create the same session object. This is good if
   * running in an app server since the sessions are elisted in the context
   * transaction. This is bad if used outside of an app server since the user may
   * be trying to create 2 different sessions to coordinate 2 different uow.
   * 
   * 
   */
  public class ActiveMQManagedConnection implements ManagedConnection, ExceptionListener { // TODO:
                                                                                              // ,
                                                                                              // DissociatableManagedConnection
                                                                                              // {
  
      private static final Logger LOG = LoggerFactory.getLogger(ActiveMQManagedConnection.class);
  
      private PrintWriter logWriter;
  
      private final ActiveMQConnection physicalConnection;
      private final TransactionContext transactionContext;
      private final List<ManagedConnectionProxy> proxyConnections = new CopyOnWriteArrayList<ManagedConnectionProxy>();
      private final List<ConnectionEventListener> listeners = new CopyOnWriteArrayList<ConnectionEventListener>();
      private final LocalAndXATransaction localAndXATransaction;
  
      private Subject subject;
      private ActiveMQConnectionRequestInfo info;
      private boolean destroyed;
  
      public ActiveMQManagedConnection(Subject subject, ActiveMQConnection physicalConnection, ActiveMQConnectionRequestInfo info) throws ResourceException {
          try {
              this.subject = subject;
              this.info = info;
              this.physicalConnection = physicalConnection;
              this.transactionContext = new TransactionContext(physicalConnection);
  
              this.localAndXATransaction = new LocalAndXATransaction(transactionContext) {
                  public void setInManagedTx(boolean inManagedTx) throws JMSException {
                      super.setInManagedTx(inManagedTx);
                      for (ManagedConnectionProxy proxy:proxyConnections) {
                          proxy.setUseSharedTxContext(inManagedTx);
                      }
                  }
              };
  
              this.transactionContext.setLocalTransactionEventListener(new LocalTransactionEventListener() {
                  public void beginEvent() {
                      fireBeginEvent();
                  }
  
                  public void commitEvent() {
                      fireCommitEvent();
                  }
  
                  public void rollbackEvent() {
                      fireRollbackEvent();
                  }
              });
  
              physicalConnection.setExceptionListener(this);
          } catch (JMSException e) {
              throw new ResourceException("Could not create a new connection: " + e.getMessage(), e);
          }
      }
  
      public boolean isInManagedTx() {
          return localAndXATransaction.isInManagedTx();
      }
  
      public static boolean matches(Object x, Object y) {
          if (x == null ^ y == null) {
              return false;
          }
          if (x != null && !x.equals(y)) {
              return false;
          }
          return true;
      }
  
      public void associate(Subject subject, ActiveMQConnectionRequestInfo info) throws JMSException {
  
          // Do we need to change the associated userid/password
          if (!matches(info.getUserName(), this.info.getUserName()) || !matches(info.getPassword(), this.info.getPassword())) {
              physicalConnection.changeUserInfo(info.getUserName(), info.getPassword());
          }
  
          // Do we need to set the clientId?
          if (info.getClientid() != null && info.getClientid().length() > 0) {
              physicalConnection.setClientID(info.getClientid());
          }
  
          this.subject = subject;
          this.info = info;
      }
  
      public Connection getPhysicalConnection() {
          return physicalConnection;
      }
  
      private void fireBeginEvent() {
          ConnectionEvent event = new ConnectionEvent(ActiveMQManagedConnection.this, ConnectionEvent.LOCAL_TRANSACTION_STARTED);
          for(ConnectionEventListener l:listeners) {
              l.localTransactionStarted(event);
          }
      }
  
      private void fireCommitEvent() {
          ConnectionEvent event = new ConnectionEvent(ActiveMQManagedConnection.this, ConnectionEvent.LOCAL_TRANSACTION_COMMITTED);
          for(ConnectionEventListener l:listeners) {
              l.localTransactionCommitted(event);
          }
      }
  
      private void fireRollbackEvent() {
          ConnectionEvent event = new ConnectionEvent(ActiveMQManagedConnection.this, ConnectionEvent.LOCAL_TRANSACTION_ROLLEDBACK);
          for(ConnectionEventListener l:listeners) {
              l.localTransactionRolledback(event);
          }
      }
  
      private void fireCloseEvent(ManagedConnectionProxy proxy) {
          ConnectionEvent event = new ConnectionEvent(ActiveMQManagedConnection.this, ConnectionEvent.CONNECTION_CLOSED);
          event.setConnectionHandle(proxy);
  
          for(ConnectionEventListener l:listeners) {
              l.connectionClosed(event);
          }
      }
  
      private void fireErrorOccurredEvent(Exception error) {
          ConnectionEvent event = new ConnectionEvent(ActiveMQManagedConnection.this, ConnectionEvent.CONNECTION_ERROR_OCCURRED, error);
          for(ConnectionEventListener l:listeners) {
              l.connectionErrorOccurred(event);
          }
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#getConnection(javax.security.auth.Subject,
       *      javax.resource.spi.ConnectionRequestInfo)
       */
      public Object getConnection(Subject subject, ConnectionRequestInfo info) throws ResourceException {
          ManagedConnectionProxy proxy = new ManagedConnectionProxy(this, info);
          proxyConnections.add(proxy);
          return proxy;
      }
  
      private boolean isDestroyed() {
          return destroyed;
      }
  
      /**
       * Close down the physical connection to the server.
       * 
       * @see javax.resource.spi.ManagedConnection#destroy()
       */
      public void destroy() throws ResourceException {
          // Have we already been destroyed??
          if (isDestroyed()) {
              return;
          }
  
          try {
              cleanup();
          } finally {
              try {
                  physicalConnection.close();
                  destroyed = true;
              } catch (JMSException e) {
                  LOG.trace("Error occurred during close of a JMS connection.", e);
              }
          }
      }
  
      /**
       * Cleans up all proxy handles attached to this physical connection so that
       * they cannot be used anymore.
       * 
       * @see javax.resource.spi.ManagedConnection#cleanup()
       */
      public void cleanup() throws ResourceException {
  
          // Have we already been destroyed??
          if (isDestroyed()) {
              return;
          }
  
          for (ManagedConnectionProxy proxy:proxyConnections) {
              proxy.cleanup();
          }
          proxyConnections.clear();
  
          try {
              physicalConnection.doCleanup(physicalConnection.isUserSpecifiedClientID());
          } catch (JMSException e) {
              throw new ResourceException("Could not cleanup the ActiveMQ connection: " + e, e);
          } finally {
              // defer transaction cleanup till after close so that close is aware of the current tx
              localAndXATransaction.cleanup();
          }
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#associateConnection(java.lang.Object)
       */
      public void associateConnection(Object connection) throws ResourceException {
          if (connection instanceof ManagedConnectionProxy) {
              ManagedConnectionProxy proxy = (ManagedConnectionProxy)connection;
              proxyConnections.add(proxy);
          } else {
              throw new ResourceException("Not supported : associating connection instance of " + connection.getClass().getName());
          }
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#addConnectionEventListener(javax.resource.spi.ConnectionEventListener)
       */
      public void addConnectionEventListener(ConnectionEventListener listener) {
          listeners.add(listener);
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#removeConnectionEventListener(javax.resource.spi.ConnectionEventListener)
       */
      public void removeConnectionEventListener(ConnectionEventListener listener) {
          listeners.remove(listener);
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#getXAResource()
       */
      public XAResource getXAResource() throws ResourceException {
          return localAndXATransaction;
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#getLocalTransaction()
       */
      public LocalTransaction getLocalTransaction() throws ResourceException {
          return localAndXATransaction;
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#getMetaData()
       */
      public ManagedConnectionMetaData getMetaData() throws ResourceException {
          return new ManagedConnectionMetaData() {
  
              public String getEISProductName() throws ResourceException {
                  if (physicalConnection == null) {
                      throw new ResourceException("Not connected.");
                  }
                  try {
                      return physicalConnection.getMetaData().getJMSProviderName();
                  } catch (JMSException e) {
                      throw new ResourceException("Error accessing provider.", e);
                  }
              }
  
              public String getEISProductVersion() throws ResourceException {
                  if (physicalConnection == null) {
                      throw new ResourceException("Not connected.");
                  }
                  try {
                      return physicalConnection.getMetaData().getProviderVersion();
                  } catch (JMSException e) {
                      throw new ResourceException("Error accessing provider.", e);
                  }
              }
  
              public int getMaxConnections() throws ResourceException {
                  if (physicalConnection == null) {
                      throw new ResourceException("Not connected.");
                  }
                  return Integer.MAX_VALUE;
              }
  
              public String getUserName() throws ResourceException {
                  if (physicalConnection == null) {
                      throw new ResourceException("Not connected.");
                  }
                  try {
                      return physicalConnection.getClientID();
                  } catch (JMSException e) {
                      throw new ResourceException("Error accessing provider.", e);
                  }
              }
          };
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#setLogWriter(java.io.PrintWriter)
       */
      public void setLogWriter(PrintWriter logWriter) throws ResourceException {
          this.logWriter = logWriter;
      }
  
      /**
       * @see javax.resource.spi.ManagedConnection#getLogWriter()
       */
      public PrintWriter getLogWriter() throws ResourceException {
          return logWriter;
      }
  
      /**
       * @param subject subject to match
       * @param info cri to match
       * @return whether the subject and cri match sufficiently to allow using this connection under the new circumstances
       */
      public boolean matches(Subject subject, ConnectionRequestInfo info) {
          // Check to see if it is our info class
          if (info == null) {
              return false;
          }
          if (info.getClass() != ActiveMQConnectionRequestInfo.class) {
              return false;
          }
  
          // Do the subjects match?
          if (subject == null ^ this.subject == null) {
              return false;
          }
          if (subject != null && !subject.equals(this.subject)) {
              return false;
          }
  
          // Does the info match?
          return info.equals(this.info);
      }
  
      /**
       * When a proxy is closed this cleans up the proxy and notifies the
       * ConnectionEventListeners that a connection closed.
       * 
       * @param proxy
       */
      public void proxyClosedEvent(ManagedConnectionProxy proxy) {
          proxyConnections.remove(proxy);
          proxy.cleanup();
          fireCloseEvent(proxy);
      }
  
      public void onException(JMSException e) {
          LOG.warn("Connection failed: " + e);
          LOG.debug("Cause: ", e);
  
          for (ManagedConnectionProxy proxy:proxyConnections) {
              proxy.onException(e);
          }
          // Let the container know that the error occurred.
          fireErrorOccurredEvent(e);
      }
  
      /**
       * @return Returns the transactionContext.
       */
      public TransactionContext getTransactionContext() {
          return transactionContext;
      }
  
      @Override
      public String toString() {
          return "[" + super.toString() + "," + physicalConnection +"]";
      }
  
  }
  