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

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;

import javax.transaction.xa.XAException;
import javax.transaction.xa.XAResource;
import javax.transaction.xa.Xid;

import org.apache.qpid.dtx.XidImpl;
import org.apache.qpid.transport.DtxXaStatus;
import org.apache.qpid.transport.ExecutionErrorCode;
import org.apache.qpid.transport.Future;
import org.apache.qpid.transport.Option;
import org.apache.qpid.transport.RecoverResult;
import org.apache.qpid.transport.SessionException;
import org.apache.qpid.transport.XaResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * This is an implementation of javax.njms.XAResource.
 */
public class XAResourceImpl implements AMQXAResource
{
    /**
     * this XAResourceImpl's logger
     */
    private static final Logger _logger = LoggerFactory.getLogger(XAResourceImpl.class);

    /**
     * Reference to the associated XASession
     */
    private XASessionImpl _xaSession = null;

    /**
     * The XID of this resource
     */
    private Xid _xid;

    /**
     * The time for this resource
     */
    private int _timeout;

    //--- constructor

    private List<XAResource> _siblings = new ArrayList<XAResource>();

    /**
     * Create an XAResource associated with a XASession
     *
     * @param xaSession The session XAresource
     */
    protected XAResourceImpl(XASessionImpl xaSession)
    {
        _xaSession = xaSession;
    }

    //--- The XAResource
    /**
     * Commits the global transaction specified by xid.
     *
     * @param xid A global transaction identifier
     * @param b   If true, use a one-phase commit protocol to commit the work done on behalf of xid.
     * @throws XAException An error has occurred. An error has occurred. Possible XAExceptions are XA_HEURHAZ,
     *                     XA_HEURCOM, XA_HEURRB, XA_HEURMIX, XAER_RMERR, XAER_RMFAIL, XAER_NOTA, XAER_INVAL, or XAER_PROTO.
     */
    public void commit(Xid xid, boolean b) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("commit tx branch with xid: {} ", xid);
        }
        Future<XaResult> future =
                _xaSession.getQpidSession().dtxCommit(convertXid(xid), b ? Option.ONE_PHASE : Option.NONE);

        // now wait on the future for the result
        XaResult result = null;
        try
        {
            result = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr(e.getException().getErrorCode());
        }
        finally
        {
            _xid = null;
        }
        checkStatus(result.getStatus());
    }

    /**
     * Ends the work performed on behalf of a transaction branch.
     * The resource manager disassociates the XA resource from the transaction branch specified
     * and lets the transaction complete.
     * <ul>
     * <li> If TMSUSPEND is specified in the flags, the transaction branch is temporarily suspended in an incomplete state.
     * The transaction context is in a suspended state and must be resumed via the start method with TMRESUME specified.
     * <li> If TMFAIL is specified, the portion of work has failed. The resource manager may mark the transaction as rollback-only
     * <li> If TMSUCCESS is specified, the portion of work has completed successfully.
     * </ul>
     *
     * @param xid  A global transaction identifier that is the same as the identifier used previously in the start method
     * @param flag One of TMSUCCESS, TMFAIL, or TMSUSPEND.
     * @throws XAException An error has occurred. An error has occurred. Possible XAException values are XAER_RMERR,
     *                     XAER_RMFAILED, XAER_NOTA, XAER_INVAL, XAER_PROTO, or XA_RB*.
     */
    public void end(Xid xid, int flag) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("end tx branch with xid: {}", xid);
        }
        switch (flag)
        {
            case(XAResource.TMSUCCESS):
                break;
            case(XAResource.TMFAIL):
                break;
            case(XAResource.TMSUSPEND):
                break;
            default:
                 throw new XAException(XAException.XAER_INVAL);
        }
        _xaSession.flushAcknowledgments();
        Future<XaResult> future = _xaSession.getQpidSession()
                .dtxEnd(convertXid(xid),
                        flag == XAResource.TMFAIL ? Option.FAIL : Option.NONE,
                        flag == XAResource.TMSUSPEND ? Option.SUSPEND : Option.NONE);
        // now wait on the future for the result
        XaResult result = null;
        try
        {
            result = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr(e.getException().getErrorCode());
        }

        checkStatus(result.getStatus());

        if(_logger.isDebugEnabled())
        {
            _logger.debug("Calling end for " + _siblings.size() + " XAResource siblings");
        }

        for(XAResource sibling: _siblings)
        {

            sibling.end(xid, flag);
        }

        _siblings.clear();
    }


    /**
     * Tells the resource manager to forget about a heuristically completed transaction branch.
     *
     * @param xid String(xid.getGlobalTransactionId() A global transaction identifier
     * @throws XAException An error has occurred. Possible exception values are XAER_RMERR, XAER_RMFAIL,
     *                     XAER_NOTA, XAER_INVAL, or XAER_PROTO.
     */
    public void forget(Xid xid) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("forget tx branch with xid: {}", xid);
        }
        _xaSession.getQpidSession().dtxForget(convertXid(xid));
        try
        {
            _xaSession.getQpidSession().sync();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr(e.getException().getErrorCode());
        }
        finally
        {
            _xid = null;
        }
    }


    /**
     * Obtains the current transaction timeout value set for this XAResource instance.
     * If XAResource.setTransactionTimeout was not used prior to invoking this method,
     * the return value is the default timeout i.e. 0;
     *
     * @return The transaction timeout value in seconds.
     * @throws XAException An error has occurred. Possible exception values are XAER_RMERR, XAER_RMFAIL.
     */
    public int getTransactionTimeout() throws XAException
    {
        return _timeout;
    }

    /**
     * This method is called to determine if the resource manager instance represented
     * by the target object is the same as the resouce manager instance represented by
     * the parameter xaResource.
     *
     * @param xaResource An XAResource object whose resource manager instance is to
     *                   be compared with the resource manager instance of the target object
     * @return <code>true</code> if it's the same RM instance; otherwise <code>false</code>.
     * @throws XAException An error has occurred. Possible exception values are XAER_RMERR, XAER_RMFAIL.
     */
    public boolean isSameRM(XAResource xaResource) throws XAException
    {
        if(this == xaResource)
        {
            return true;
        }

        if(!(xaResource instanceof AMQXAResource))
        {
            return false;
        }

        String myUUID = getBrokerUUID();
        String otherUUID = ((AMQXAResource)xaResource).getBrokerUUID();

        if(_logger.isDebugEnabled())
        {
            _logger.debug("Comparing my UUID " + myUUID + " with other UUID " + otherUUID);
        }

        boolean isSameRm = (myUUID != null && otherUUID != null && myUUID.equals(otherUUID));

        if(isSameRm)
        {
            if(_logger.isDebugEnabled())
            {
                _logger.debug("XAResource " + xaResource + " is from the ResourceManager. Adding XAResource as sibling for AMQP protocol support. ");
            }
            _siblings.add(xaResource);
        }

        return isSameRm;

    }

    /**
     * Prepare for a transaction commit of the transaction specified in <code>Xid</code>.
     *
     * @param xid A global transaction identifier.
     * @return A value indicating the resource manager's vote on the outcome of the transaction.
     *         The possible values are: XA_RDONLY or XA_OK.
     * @throws XAException An error has occurred. Possible exception values are: XAER_RMERR or XAER_NOTA
     */
    public int prepare(Xid xid) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("prepare {}", xid);
        }
        Future<XaResult> future = _xaSession.getQpidSession().dtxPrepare(convertXid(xid));
        XaResult result = null;
        try
        {
            result = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr(e.getException().getErrorCode());
        }
        DtxXaStatus status = result.getStatus();
        int outcome = XAResource.XA_OK;
        switch (status)
        {
            case XA_OK:
                break;
            case XA_RDONLY:
                outcome = XAResource.XA_RDONLY;
                break;
            default:
                checkStatus(status);
        }
        return outcome;
    }

    /**
     * Obtains a list of prepared transaction branches.
     * <p>
     * The transaction manager calls this method during recovery to obtain the list of transaction branches
     * that are currently in prepared or heuristically completed states.
     *
     * @param flag One of TMSTARTRSCAN, TMENDRSCAN, TMNOFLAGS.
     *             TMNOFLAGS must be used when no other flags are set in the parameter.
     * @return zero or more XIDs of the transaction branches that are currently in a prepared or heuristically
     *         completed state.
     * @throws XAException An error has occurred. Possible value is XAER_INVAL.
     */
    public Xid[] recover(int flag) throws XAException
    {
        // the flag is ignored
        Future<RecoverResult> future = _xaSession.getQpidSession().dtxRecover();
        RecoverResult res = null;
        try
        {
            res = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr( e.getException().getErrorCode());
        }
        Xid[] result =  new Xid[res.getInDoubt() != null ? res.getInDoubt().size() : 0];
        if(result.length != 0)
        {
            int i = 0;
            for (Object obj : res.getInDoubt())
            {
                org.apache.qpid.transport.Xid xid = (org.apache.qpid.transport.Xid) obj;
                result[i] = new XidImpl(xid.getBranchId(), (int) xid.getFormat(), xid.getGlobalId());
                i++;
            }
        }
        return result;
    }

    /**
     * Informs the resource manager to roll back work done on behalf of a transaction branch
     *
     * @param xid A global transaction identifier.
     * @throws XAException An error has occurred.
     */
    public void rollback(Xid xid) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("rollback tx branch with xid: {}", xid);
        }

        Future<XaResult> future = _xaSession.getQpidSession().dtxRollback(convertXid(xid));
        // now wait on the future for the result
        XaResult result = null;
        try
        {
            result = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr( e.getException().getErrorCode());
        }
        finally
        {
            _xid = null;
        }
        checkStatus(result.getStatus());
    }

    /**
     * Sets the current transaction timeout value for this XAResource instance.
     * Once set, this timeout value is effective until setTransactionTimeout is
     * invoked again with a different value.
     * To reset the timeout value to the default value used by the resource manager, set the value to zero.
     *
     * @param timeout The transaction timeout value in seconds.
     * @return true if transaction timeout value is set successfully; otherwise false.
     * @throws XAException An error has occurred. Possible exception values are XAER_RMERR, XAER_RMFAIL, or XAER_INVAL.
     */
    public boolean setTransactionTimeout(int timeout) throws XAException
    {
        _timeout = timeout;
        if (timeout != _timeout && _xid != null)
        {
            setDtxTimeout(_timeout);
        }
        return true;
    }

    private void setDtxTimeout(int timeout) throws XAException
    {
        _xaSession.getQpidSession()
                    .dtxSetTimeout(XidImpl.convert(_xid), timeout);
    }

    /**
     * Starts work on behalf of a transaction branch specified in xid.
     * <ul>
     * <li> If TMJOIN is specified, an exception is thrown as it is not supported
     * <li> If TMRESUME is specified, the start applies to resuming a suspended transaction specified in the parameter xid.
     * <li> If neither TMJOIN nor TMRESUME is specified and the transaction specified by xid has previously been seen by the
     * resource manager, the resource manager throws the XAException exception with XAER_DUPID error code.
     * </ul>
     *
     * @param xid  A global transaction identifier to be associated with the resource
     * @param flag One of TMNOFLAGS, TMJOIN, or TMRESUME
     * @throws XAException An error has occurred. Possible exceptions
     *                     are XA_RB*, XAER_RMERR, XAER_RMFAIL, XAER_DUPID, XAER_OUTSIDE, XAER_NOTA, XAER_INVAL, or XAER_PROTO.
     */
    public void start(Xid xid, int flag) throws XAException
    {
        if (_logger.isDebugEnabled())
        {
            _logger.debug("start tx branch with xid: {}", xid);
        }
        switch (flag)
        {
            case(XAResource.TMNOFLAGS):
                break;
            case(XAResource.TMJOIN):
                break;
            case(XAResource.TMRESUME):
                break;
            default:
                 throw new XAException(XAException.XAER_INVAL);
        }
        Future<XaResult> future = _xaSession.getQpidSession()
                .dtxStart(convertXid(xid),
                        flag == XAResource.TMJOIN ? Option.JOIN : Option.NONE,
                        flag == XAResource.TMRESUME ? Option.RESUME : Option.NONE);
        // now wait on the future for the result
        XaResult result = null;
        try
        {
            result = future.get();
        }
        catch (SessionException e)
        {
            // we need to restore the qpid session that has been closed
            _xaSession.createSession();
            convertExecutionErrorToXAErr(e.getException().getErrorCode());
            // TODO: The amqp spec does not allow to make the difference
            // between an already known XID and a wrong arguments (join and resume are set)
            // TODO: make sure amqp addresses that
        }
        checkStatus(result.getStatus());
        _xid = xid;
        if (_timeout > 0)
        {
            setDtxTimeout(_timeout);
        }

        for(XAResource sibling: _siblings)
        {
            sibling.start(xid, flag);
        }
    }

    /**
     * Is this resource currently enlisted in a transaction?
     *
     * @return true if the resource is associated with a transaction, false otherwise.
     */
    public boolean isEnlisted()
    {
        return (_xid != null) ;
    }

    //------------------------------------------------------------------------
    // Private methods
    //------------------------------------------------------------------------

    /**
     * Check xa method outcome and, when required, convert the status into the corresponding xa exception
     * @param status method status code
     * @throws XAException corresponding XA Exception when required
     */
    private void checkStatus(DtxXaStatus status) throws XAException
    {
        switch (status)
        {
            case XA_OK:
                // Do nothing this ok
                break;
            case XA_RBROLLBACK:
                // The tx has been rolled back for an unspecified reason.
                throw new XAException(XAException.XA_RBROLLBACK);
            case XA_RBTIMEOUT:
                // The transaction branch took too long.
                throw new XAException(XAException.XA_RBTIMEOUT);
            case XA_HEURHAZ:
                // The transaction branch may have been heuristically completed.
                throw new XAException(XAException.XA_HEURHAZ);
            case XA_HEURCOM:
                // The transaction branch has been heuristically committed.
                throw new XAException(XAException.XA_HEURCOM);
            case XA_HEURRB:
                // The transaction branch has been heuristically rolled back.
                throw new XAException(XAException.XA_HEURRB);
            case XA_HEURMIX:
                // The transaction branch has been heuristically committed and rolled back.
                throw new XAException(XAException.XA_HEURMIX);
            case XA_RDONLY:
                // The transaction branch was read-only and has been committed.
                throw new XAException(XAException.XA_RDONLY);
            default:
                // this should not happen
                if (_logger.isDebugEnabled())
                {
                    _logger.debug("got unexpected status value: {}", status);
                }
                //A resource manager error has occured in the transaction branch.
                throw new XAException(XAException.XAER_RMERR);
        }
    }

    /**
     * Convert execution error to xa exception.
     * @param error the execution error code
     * @throws XAException
     */
    private void convertExecutionErrorToXAErr(ExecutionErrorCode error) throws XAException
    {
        switch (error)
        {
            case NOT_ALLOWED:
                // The XID already exists.
                throw new XAException(XAException.XAER_DUPID);
            case NOT_FOUND:
                // The XID is not valid.
                try
                {
                   throw new XAException(XAException.XAER_NOTA);
                }
                catch (XAException e)
                {
                    _logger.error(e.getMessage(), e);
                    throw e;
                }
            case ILLEGAL_STATE:
                // Routine was invoked in an inproper context.
                throw new XAException(XAException.XAER_PROTO);
            case NOT_IMPLEMENTED:
                // the command is not implemented
                throw new XAException(XAException.XAER_RMERR);
            case COMMAND_INVALID:
                // Invalid call
                throw new XAException(XAException.XAER_INVAL);
            default:
                // this should not happen
                if (_logger.isDebugEnabled())
                {
                    _logger.debug("Got unexpected error: " + error);
                }
                //A resource manager error has occured in the transaction branch.
                throw new XAException(XAException.XAER_RMERR);
        }
    }

    /**
     * convert a generic xid into qpid format
     * @param xid xid to be converted
     * @return the qpid formated xid
     * @throws XAException when xid is null
     */
    private org.apache.qpid.transport.Xid convertXid(Xid xid) throws XAException
    {
        if (xid == null)
        {
            // Invalid arguments were given.
            throw new XAException(XAException.XAER_INVAL);
        }
        return XidImpl.convert(xid);
    }

    public String getBrokerUUID()
    {
        return ((AMQSession_0_10)_xaSession).getAMQConnection().getBrokerUUID();
    }

    public List<XAResource> getSiblings()
    {
        return Collections.unmodifiableList(_siblings);
    }
}