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

import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.jms.Session;
import javax.jms.XASession;
import javax.resource.ResourceException;
import javax.resource.spi.ActivationSpec;
import javax.resource.spi.BootstrapContext;
import javax.resource.spi.ResourceAdapter;
import javax.resource.spi.ResourceAdapterInternalException;
import javax.resource.spi.endpoint.MessageEndpointFactory;
import javax.resource.spi.work.WorkManager;
import javax.transaction.TransactionManager;
import javax.transaction.xa.XAResource;

import org.apache.qpid.client.AMQConnection;
import org.apache.qpid.client.AMQConnectionFactory;
import org.apache.qpid.client.AMQConnectionURL;
import org.apache.qpid.client.XAConnectionImpl;
import org.apache.qpid.ra.inflow.QpidActivation;
import org.apache.qpid.ra.inflow.QpidActivationSpec;
import org.apache.qpid.url.URLSyntaxException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The resource adapter for Qpid
 *
 */
public class QpidResourceAdapter implements ResourceAdapter, Serializable
{
   private static final long serialVersionUID = -2446231446818098726L;

   private static final transient Logger _log = LoggerFactory.getLogger(QpidResourceAdapter.class);

   private BootstrapContext _ctx;

   private final QpidRAProperties _raProperties;

   private final AtomicBoolean _configured;

   private final Map<ActivationSpec, QpidActivation> _activations;

   private AMQConnectionFactory _defaultAMQConnectionFactory;

   private TransactionManager _tm;

   public QpidResourceAdapter()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("constructor()");
      }

      _raProperties = new QpidRAProperties();
      _configured = new AtomicBoolean(false);
      _activations = new ConcurrentHashMap<ActivationSpec, QpidActivation>();
   }

   public TransactionManager getTM()
   {
      return _tm;
   }

   /**
    * Endpoint activation
    *
    * @param endpointFactory The endpoint factory
    * @param spec            The activation spec
    * @throws ResourceException Thrown if an error occurs
    */
   public void endpointActivation(final MessageEndpointFactory endpointFactory, final ActivationSpec spec) throws ResourceException
   {
      if (!_configured.getAndSet(true))
      {
         try
         {
            setup();
         }
         catch (QpidRAException e)
         {
            throw new ResourceException("Unable to create activation", e);
         }
      }
      if (_log.isTraceEnabled())
      {
         _log.trace("endpointActivation(" + endpointFactory + ", " + spec + ")");
      }

      QpidActivation activation = new QpidActivation(this, endpointFactory, (QpidActivationSpec)spec);
      _activations.put(spec, activation);
      activation.start();
   }

   /**
    * Endpoint deactivation
    *
    * @param endpointFactory The endpoint factory
    * @param spec            The activation spec
    */
   public void endpointDeactivation(final MessageEndpointFactory endpointFactory, final ActivationSpec spec)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("endpointDeactivation(" + endpointFactory + ", " + spec + ")");
      }

      QpidActivation activation = _activations.remove(spec);
      if (activation != null)
      {
         activation.stop();
      }
   }

   /**
    * Get XA resources
    *
    * @param specs The activation specs
    * @return The XA resources
    * @throws ResourceException Thrown if an error occurs or unsupported
    */
   public XAResource[] getXAResources(final ActivationSpec[] specs) throws ResourceException
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getXAResources(" + specs + ")");
      }

      return null;
   }

   /**
    * Start
    *
    * @param ctx The bootstrap context
    * @throws ResourceAdapterInternalException
    *          Thrown if an error occurs
    */
   public void start(final BootstrapContext ctx) throws ResourceAdapterInternalException
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("start(" + ctx + ")");
      }

      locateTM();

      this._ctx = ctx;

      _log.info("Qpid resource adapter started");
   }

   /**
    * Stop
    */
   public void stop()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("stop()");
      }

      for (Map.Entry<ActivationSpec, QpidActivation> entry : _activations.entrySet())
      {
         try
         {
            entry.getValue().stop();
         }
         catch (Exception ignored)
         {
            _log.debug("Ignored", ignored);
         }
      }

      _activations.clear();

      _log.info("Qpid resource adapter stopped");
   }


   /**
    * Get the client ID
    *
    * @return The value
    */
   public String getClientId()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getClientID()");
      }

      return _raProperties.getClientId();
   }

   /**
    * Set the client ID
    *
    * @param clientID The client id
    */
   public void setClientId(final String clientID)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setClientID(" + clientID + ")");
      }

      _raProperties.setClientId(clientID);
   }

   /**
    * Get the host
    *
    * @return The value
    */
   public String getHost()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getHost()");
      }

      return _raProperties.getHost();
   }

   /**
    * Set the host
    *
    * @param host The host
    */
   public void setHost(final String host)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setHost(" + host + ")");
      }

      _raProperties.setHost(host);
   }

   /**
    * Get the port
    *
    * @return The value
    */
   public Integer getPort()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getPort()");
      }

      return _raProperties.getPort();
   }

   /**
    * Set the client ID
    *
    * @param port The port
    */
   public void setPort(final Integer port)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setPort(" + port + ")");
      }

      _raProperties.setPort(port);
   }

   /**
    * Get the connection url
    *
    * @return The value
    */
   public String getPath()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getPath()");
      }

      return _raProperties.getPath();
   }

   /**
    * Set the client ID
    *
    * @param path The path
    */
   public void setPath(final String path)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setPath(" + path + ")");
      }

      _raProperties.setPath(path);
   }

   public String getUserName()
   {
       return _raProperties.getUserName();
   }

   public void setUserName(String userName)
   {
      _raProperties.setUserName(userName);
   }

   public String getPassword()
   {
       return _raProperties.getPassword();
   }

   public void setPassword(String password)
   {
       _raProperties.setPassword(password);
   }

   /**
    * Get the connection url
    *
    * @return The value
    */
   public String getConnectionURL()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getConnectionURL()");
      }

      return _raProperties.getConnectionURL();
   }

   /**
    * Set the client ID
    *
    * @param connectionURL The connection url
    */
   public void setConnectionURL(final String connectionURL)
   {
      _raProperties.setConnectionURL(connectionURL);
   }

   /**
    * Get the transaction manager locator class
    *
    * @return The value
    */
   public String getTransactionManagerLocatorClass()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getTransactionManagerLocatorClass()");
      }

      return _raProperties.getTransactionManagerLocatorClass();
   }

   /**
    * Set the transaction manager locator class
    *
    * @param locator The transaction manager locator class
    */
   public void setTransactionManagerLocatorClass(final String locator)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setTransactionManagerLocatorClass(" + locator + ")");
      }

      _raProperties.setTransactionManagerLocatorClass(locator);
   }

   /**
    * Get the transaction manager locator method
    *
    * @return The value
    */
   public String getTransactionManagerLocatorMethod()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getTransactionManagerLocatorMethod()");
      }

      return _raProperties.getTransactionManagerLocatorMethod();
   }

   /**
    * Set the transaction manager locator method
    *
    * @param method The transaction manager locator method
    */
   public void setTransactionManagerLocatorMethod(final String method)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setTransactionManagerLocatorMethod(" + method + ")");
      }

      _raProperties.setTransactionManagerLocatorMethod(method);
   }

   /**
    * Get the use XA flag
    *
    * @return The value
    */
   public Boolean isUseLocalTx()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getUseLocalTx()");
      }

      return _raProperties.isUseLocalTx();
   }

   /**
    * Set the use XA flag
    *
    * @param localTx The value
    */
   public void setUseLocalTx(final Boolean localTx)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setUseLocalTx(" + localTx + ")");
      }

      _raProperties.setUseLocalTx(localTx);
   }

   public Integer getSetupAttempts()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getSetupAttempts()");
      }
      return _raProperties.getSetupAttempts();
   }

   public void setSetupAttempts(Integer setupAttempts)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setSetupAttempts(" + setupAttempts + ")");
      }
      _raProperties.setSetupAttempts(setupAttempts);
   }

   public Long getSetupInterval()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getSetupInterval()");
      }
      return _raProperties.getSetupInterval();
   }

   public void setSetupInterval(Long interval)
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("setSetupInterval(" + interval + ")");
      }
      _raProperties.setSetupInterval(interval);
   }
   
   public Boolean isUseConnectionPerHandler()
   {
       if (_log.isTraceEnabled())
       {
          _log.trace("isConnectionPerHandler()");
       }
       
       return _raProperties.isUseConnectionPerHandler();
   }

   public void setUseConnectionPerHandler(Boolean connectionPerHandler)
   {
       if (_log.isTraceEnabled())
       {
          _log.trace("setConnectionPerHandler(" + connectionPerHandler + ")");
       }  
       
       _raProperties.setUseConnectionPerHandler(connectionPerHandler);
   }
   
   /**
    * Indicates whether some other object is "equal to" this one.
    *
    * @param obj Object with which to compare
    * @return True if this object is the same as the obj argument; false otherwise.
    */
   public boolean equals(final Object obj)
   {
      if (obj == null)
      {
         return false;
      }

      if (obj instanceof QpidResourceAdapter)
      {
         return _raProperties.equals(((QpidResourceAdapter)obj).getProperties());
      }
      else
      {
         return false;
      }
   }

   /**
    * Return the hash code for the object
    *
    * @return The hash code
    */
   public int hashCode()
   {
      return _raProperties.hashCode();
   }

   /**
    * Get the work manager
    *
    * @return The manager
    */
   public WorkManager getWorkManager()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getWorkManager()");
      }

      if (_ctx == null)
      {
         return null;
      }

      return _ctx.getWorkManager();
   }

   public XASession createXASession(final XAConnectionImpl connection)
      throws Exception
   {
      final XASession result = connection.createXASession() ;
      if (_log.isDebugEnabled())
      {
         _log.debug("Using session " + Util.asString(result));
      }
      return result ;
   }

   public Session createSession(final AMQConnection connection,
                                      final int ackMode,
                                      final boolean useLocalTx,
                                      final Integer prefetchLow,
                                      final Integer prefetchHigh) throws Exception
   {
      Session result;

      if (prefetchLow == null)
      {
         result = connection.createSession(useLocalTx, ackMode) ;
      }
      else if (prefetchHigh == null)
      {
         result = connection.createSession(useLocalTx, ackMode, prefetchLow) ;
      }
      else
      {
         result = connection.createSession(useLocalTx, ackMode, prefetchHigh, prefetchLow) ;
      }

      if (_log.isDebugEnabled())
      {
         _log.debug("Using session " + Util.asString(result));
      }

      return result;

   }

   /**
    * Get the resource adapter properties
    *
    * @return The properties
    */
   protected QpidRAProperties getProperties()
   {
      if (_log.isTraceEnabled())
      {
         _log.trace("getProperties()");
      }

      return _raProperties;
   }

   /**
    * Setup the factory
    */
   protected void setup() throws QpidRAException
   {
      _defaultAMQConnectionFactory = createAMQConnectionFactory(_raProperties);
   }


   public AMQConnectionFactory getDefaultAMQConnectionFactory() throws ResourceException
   {
      if (!_configured.getAndSet(true))
      {
         try
         {
            setup();
         }
         catch (QpidRAException e)
         {
            throw new ResourceException("Unable to create activation", e);
         }
      }
      return _defaultAMQConnectionFactory;
   }

   public AMQConnectionFactory createAMQConnectionFactory(final ConnectionFactoryProperties overrideProperties)
      throws QpidRAException
   {
      try
      {
         return createFactory(overrideProperties);
      }
      catch (final URLSyntaxException urlse)
      {
         throw new QpidRAException("Unexpected exception creating connection factory", urlse) ;
      }
   }

   public Map<String, Object> overrideConnectionParameters(final Map<String, Object> connectionParams,
                                                           final Map<String, Object> overrideConnectionParams)
   {
      Map<String, Object> map = new HashMap<String, Object>();

      if(connectionParams != null)
      {
         map.putAll(connectionParams);
      }
      if(overrideConnectionParams != null)
      {
         for (Map.Entry<String, Object> stringObjectEntry : overrideConnectionParams.entrySet())
         {
            map.put(stringObjectEntry.getKey(), stringObjectEntry.getValue());
         }
      }
      return map;
   }

   private void locateTM() throws ResourceAdapterInternalException
   {
      if(_raProperties.getTransactionManagerLocatorClass() != null 
                      && _raProperties.getTransactionManagerLocatorMethod() != null)
      {

          String locatorClasses[] = _raProperties.getTransactionManagerLocatorClass().split(";");
          String locatorMethods[] = _raProperties.getTransactionManagerLocatorMethod().split(";");

          for (int i = 0 ; i < locatorClasses.length; i++)
          {
              _tm = Util.locateTM(locatorClasses[i], locatorMethods[i]);
              if (_tm != null)
              {
                  break;
              }
          }


      }

      if (_tm == null)
      {
         _log.error("It was not possible to locate javax.transaction.TransactionManager via the RA properties TransactionManagerLocatorClass and TransactionManagerLocatorMethod");
         throw new ResourceAdapterInternalException("Could not locate javax.transaction.TransactionManager");
      }
      else
      {
         if (_log.isDebugEnabled())
         {
            _log.debug("TM located = " + _tm);
         }
      }
   }


   private AMQConnectionFactory createFactory(final ConnectionFactoryProperties overrideProperties)
      throws URLSyntaxException, QpidRAException
   {
      final String overrideURL = overrideProperties.getConnectionURL() ;
      final String url = overrideURL != null ? overrideURL : _raProperties.getConnectionURL() ;

      final String overrideClientID = overrideProperties.getClientId() ;
      final String clientID = (overrideClientID != null ? overrideClientID : _raProperties.getClientId()) ;

      final String overrideDefaultPassword = overrideProperties.getPassword() ;
      final String defaultPassword = (overrideDefaultPassword != null ? overrideDefaultPassword : _raProperties.getPassword()) ;

      final String overrideDefaultUsername = overrideProperties.getUserName() ;
      final String defaultUsername = (overrideDefaultUsername != null ? overrideDefaultUsername : _raProperties.getUserName()) ;

      final String overrideHost = overrideProperties.getHost() ;
      final String host = (overrideHost != null ? overrideHost : _raProperties.getHost()) ;

      final Integer overridePort = overrideProperties.getPort() ;
      final Integer port = (overridePort != null ? overridePort : _raProperties.getPort()) ;

      final String overridePath = overrideProperties.getPath() ;
      final String path = (overridePath != null ? overridePath : _raProperties.getPath()) ;

      final AMQConnectionFactory cf ;

      if (url != null)
      {
         cf = new AMQConnectionFactory(url) ;

         if (clientID != null)
         {
            cf.getConnectionURL().setClientName(clientID) ;
         }
      }
      else
      {
         // create a URL to force the connection details
         if ((host == null) || (port == null) || (path == null))
         {
            throw new QpidRAException("Configuration requires host/port/path if connectionURL is not specified") ;
         }
         final String username = (defaultUsername != null ? defaultUsername : "") ;
         final String password = (defaultPassword != null ? defaultPassword : "") ;
         final String client = (clientID != null ? clientID : "") ;

         final String newurl = AMQConnectionURL.AMQ_PROTOCOL + "://" + username +":" + password + "@" + client + "/" + path + '?' + AMQConnectionURL.OPTIONS_BROKERLIST + "='tcp://" + host + ':' + port + '\'' ;
         
         if (_log.isDebugEnabled())
         {
            _log.debug("Initialising connectionURL to " + newurl) ;
         }

         cf = new AMQConnectionFactory(newurl) ;
      }

      return cf ;
   }
}