//
//  ========================================================================
//  Copyright (c) 1995-2013 Mort Bay Consulting Pty. Ltd.
//  ------------------------------------------------------------------------
//  All rights reserved. This program and the accompanying materials
//  are made available under the terms of the Eclipse Public License v1.0
//  and Apache License v2.0 which accompanies this distribution.
//
//      The Eclipse Public License is available at
//      http://www.eclipse.org/legal/epl-v10.html
//
//      The Apache License v2.0 is available at
//      http://www.opensource.org/licenses/apache2.0.php
//
//  You may elect to redistribute this code under either of these licenses.
//  ========================================================================
//

package org.eclipse.jetty.client;

import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicMarkableReference;

import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.api.Response;
import org.eclipse.jetty.client.api.Result;
import org.eclipse.jetty.util.log.Log;
import org.eclipse.jetty.util.log.Logger;

public class HttpExchange
{
    private static final Logger LOG = Log.getLogger(HttpExchange.class);

    private final AtomicInteger complete = new AtomicInteger();
    private final CountDownLatch terminate = new CountDownLatch(2);
    private final HttpConversation conversation;
    private final HttpDestination destination;
    private final Request request;
    private final List<Response.ResponseListener> listeners;
    private final HttpResponse response;
    private volatile HttpConnection connection;
    private volatile Throwable requestFailure;
    private volatile Throwable responseFailure;

    public HttpExchange(HttpConversation conversation, HttpDestination destination, Request request, List<Response.ResponseListener> listeners)
    {
        this.conversation = conversation;
        this.destination = destination;
        this.request = request;
        this.listeners = listeners;
        this.response = new HttpResponse(request, listeners);
        conversation.getExchanges().offer(this);
        conversation.updateResponseListeners(null);
    }

    public HttpConversation getConversation()
    {
        return conversation;
    }

    public Request getRequest()
    {
        return request;
    }

    public Throwable getRequestFailure()
    {
        return requestFailure;
    }

    public List<Response.ResponseListener> getResponseListeners()
    {
        return listeners;
    }

    public HttpResponse getResponse()
    {
        return response;
    }

    public Throwable getResponseFailure()
    {
        return responseFailure;
    }

    public void setConnection(HttpConnection connection)
    {
        this.connection = connection;
    }

    public AtomicMarkableReference<Result> requestComplete(Throwable failure)
    {
        int requestSuccess = 0b0011;
        int requestFailure = 0b0001;
        return complete(failure == null ? requestSuccess : requestFailure, failure);
    }

    public AtomicMarkableReference<Result> responseComplete(Throwable failure)
    {
        if (failure == null)
        {
            int responseSuccess = 0b1100;
            return complete(responseSuccess, failure);
        }
        else
        {
            proceed(false);
            int responseFailure = 0b0100;
            return complete(responseFailure, failure);
        }
    }

    /**
     * This method needs to atomically compute whether this exchange is completed,
     * that is both request and responses are completed (either with a success or
     * a failure).
     *
     * Furthermore, this method needs to atomically compute whether the exchange
     * has completed successfully (both request and response are successful) or not.
     *
     * To do this, we use 2 bits for the request (one to indicate completion, one
     * to indicate success), and similarly for the response.
     * By using {@link AtomicInteger} to atomically sum these codes we can know
     * whether the exchange is completed and whether is successful.
     *
     * @param code the bits representing the status code for either the request or the response
     * @param failure the failure - if any - associated with the status code for either the request or the response
     * @return an AtomicMarkableReference holding whether the operation modified the
     * completion status and the {@link Result} - if any - associated with the status
     */
    private AtomicMarkableReference<Result> complete(int code, Throwable failure)
    {
        Result result = null;
        boolean modified = false;

        int current;
        while (true)
        {
            current = complete.get();
            boolean updateable = (current & code) == 0;
            if (updateable)
            {
                int candidate = current | code;
                if (!complete.compareAndSet(current, candidate))
                    continue;
                current = candidate;
                modified = true;
                if ((code & 0b01) == 0b01)
                    requestFailure = failure;
                else
                    responseFailure = failure;
                LOG.debug("{} updated", this);
            }
            break;
        }

        int completed = 0b0101;
        if ((current & completed) == completed)
        {
            if (modified)
            {
                // Request and response completed
                LOG.debug("{} complete", this);
                conversation.complete();
            }
            result = new Result(getRequest(), getRequestFailure(), getResponse(), getResponseFailure());
        }

        return new AtomicMarkableReference<>(result, modified);
    }

    public boolean abort(Throwable cause)
    {
        if (destination.remove(this))
        {
            destination.abort(this, cause);
            LOG.debug("Aborted while queued {}: {}", this, cause);
            return true;
        }
        else
        {
            HttpConnection connection = this.connection;
            // If there is no connection, this exchange is already completed
            if (connection == null)
                return false;

            boolean aborted = connection.abort(cause);
            LOG.debug("Aborted while active ({}) {}: {}", aborted, this, cause);
            return aborted;
        }
    }

    public void resetResponse(boolean success)
    {
        int responseSuccess = 0b1100;
        int responseFailure = 0b0100;
        int code = success ? responseSuccess : responseFailure;
        complete.addAndGet(-code);
    }

    public void proceed(boolean proceed)
    {
        HttpConnection connection = this.connection;
        if (connection != null)
            connection.proceed(proceed);
    }

    public void terminateRequest()
    {
        terminate.countDown();
    }

    public void terminateResponse()
    {
        terminate.countDown();
    }

    public void awaitTermination()
    {
        try
        {
            terminate.await();
        }
        catch (InterruptedException x)
        {
            LOG.ignore(x);
        }
    }

    @Override
    public String toString()
    {
        String padding = "0000";
        String status = Integer.toBinaryString(complete.get());
        return String.format("%s@%x status=%s%s",
                HttpExchange.class.getSimpleName(),
                hashCode(),
                padding.substring(status.length()),
                status);
    }
}
