package org.apache.lucene.replicator.http;

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

import java.io.Closeable;
import java.io.IOException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.concurrent.Callable;

import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.HttpStatus;
import org.apache.http.StatusLine;
import org.apache.http.client.HttpClient;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.conn.ClientConnectionManager;
import org.apache.http.impl.client.DefaultHttpClient;
import org.apache.http.params.HttpConnectionParams;
import org.apache.http.util.EntityUtils;
import org.apache.lucene.store.AlreadyClosedException;

/**
 * Base class for Http clients.
 * 
 * @lucene.experimental
 * */
public abstract class HttpClientBase implements Closeable {
  
  /**
   * Default connection timeout for this client, in milliseconds.
   * 
   * @see #setConnectionTimeout(int)
   */
  public static final int DEFAULT_CONNECTION_TIMEOUT = 1000;
  
  /**
   * Default socket timeout for this client, in milliseconds.
   * 
   * @see #setSoTimeout(int)
   */
  public static final int DEFAULT_SO_TIMEOUT = 60000;
  
  // TODO compression?
  
  /** The URL stting to execute requests against. */
  protected final String url;
  
  private volatile boolean closed = false;
  
  private final HttpClient httpc;
  
  /**
   * @param conMgr connection manager to use for this http client.
   *        <b>NOTE:</b>The provided {@link ClientConnectionManager} will not be
   *        {@link ClientConnectionManager#shutdown()} by this class.
   */
  protected HttpClientBase(String host, int port, String path, ClientConnectionManager conMgr) {
    url = normalizedURL(host, port, path);
    httpc = new DefaultHttpClient(conMgr);
    setConnectionTimeout(DEFAULT_CONNECTION_TIMEOUT);
    setSoTimeout(DEFAULT_SO_TIMEOUT);
  }
  
  /**
   * Set the connection timeout for this client, in milliseconds. This setting
   * is used to modify {@link HttpConnectionParams#setConnectionTimeout}.
   * 
   * @param timeout timeout to set, in millisecopnds
   */
  public void setConnectionTimeout(int timeout) {
    HttpConnectionParams.setConnectionTimeout(httpc.getParams(), timeout);
  }
  
  /**
   * Set the socket timeout for this client, in milliseconds. This setting
   * is used to modify {@link HttpConnectionParams#setSoTimeout}.
   * 
   * @param timeout timeout to set, in millisecopnds
   */
  public void setSoTimeout(int timeout) {
    HttpConnectionParams.setSoTimeout(httpc.getParams(), timeout);
  }
  
  /** Throws {@link AlreadyClosedException} if this client is already closed. */
  protected final void ensureOpen() throws AlreadyClosedException {
    if (closed) {
      throw new AlreadyClosedException("HttpClient already closed");
    }
  }
  
  /**
   * Create a URL out of the given parameters, translate an empty/null path to '/'
   */
  private static String normalizedURL(String host, int port, String path) {
    if (path == null || path.length() == 0) {
      path = "/";
    }
    return "http://" + host + ":" + port + path;
  }
  
  /**
   * <b>Internal:</b> response status after invocation, and in case or error attempt to read the 
   * exception sent by the server. 
   */
  protected void verifyStatus(HttpResponse response) throws IOException {
    StatusLine statusLine = response.getStatusLine();
    if (statusLine.getStatusCode() != HttpStatus.SC_OK) {
      throwKnownError(response, statusLine); 
    }
  }
  
  protected void throwKnownError(HttpResponse response, StatusLine statusLine) throws IOException {
    ObjectInputStream in = null;
    try {
      in = new ObjectInputStream(response.getEntity().getContent());
    } catch (Exception e) {
      // the response stream is not an exception - could be an error in servlet.init().
      throw new RuntimeException("Uknown error: " + statusLine);
    }
    
    Throwable t;
    try {
      t = (Throwable) in.readObject();
    } catch (Exception e) { 
      //not likely
      throw new RuntimeException("Failed to read exception object: " + statusLine, e);
    } finally {
      in.close();
    }
    if (t instanceof IOException) {
      throw (IOException) t;
    }
    if (t instanceof RuntimeException) {
      throw (RuntimeException) t;
    }
    throw new RuntimeException("unknown exception "+statusLine,t);
  }
  
  /**
   * <b>internal:</b> execute a request and return its result
   * The <code>params</code> argument is treated as: name1,value1,name2,value2,...
   */
  protected HttpResponse executePOST(String request, HttpEntity entity, String... params) throws IOException {
    ensureOpen();
    HttpPost m = new HttpPost(queryString(request, params));
    m.setEntity(entity);
    HttpResponse response = httpc.execute(m);
    verifyStatus(response);
    return response;
  }
  
  /**
   * <b>internal:</b> execute a request and return its result
   * The <code>params</code> argument is treated as: name1,value1,name2,value2,...
   */
  protected HttpResponse executeGET(String request, String... params) throws IOException {
    ensureOpen();
    HttpGet m = new HttpGet(queryString(request, params));
    HttpResponse response = httpc.execute(m);
    verifyStatus(response);
    return response;
  }
  
  private String queryString(String request, String... params) throws UnsupportedEncodingException {
    StringBuilder query = new StringBuilder(url).append('/').append(request).append('?');
    if (params != null) {
      for (int i = 0; i < params.length; i += 2) {
        query.append(params[i]).append('=').append(URLEncoder.encode(params[i+1], "UTF8")).append('&');
      }
    }
    return query.substring(0, query.length() - 1);
  }
  
  /** Internal utility: input stream of the provided response */
  public InputStream responseInputStream(HttpResponse response) throws IOException {
    return responseInputStream(response, false);
  }
  
  // TODO: can we simplify this Consuming !?!?!?
  /**
   * Internal utility: input stream of the provided response, which optionally 
   * consumes the response's resources when the input stream is exhausted.
   */
  public InputStream responseInputStream(HttpResponse response, boolean consume) throws IOException {
    final HttpEntity entity = response.getEntity();
    final InputStream in = entity.getContent();
    if (!consume) {
      return in;
    }
    return new InputStream() {
      private boolean consumed = false;
      @Override
      public int read() throws IOException {
        final int res = in.read();
        consume(res);
        return res;
      }
      @Override
      public void close() throws IOException {
        super.close();
        consume(-1);
      }
      @Override
      public int read(byte[] b) throws IOException {
        final int res = super.read(b);
        consume(res);
        return res;
      }
      @Override
      public int read(byte[] b, int off, int len) throws IOException {
        final int res = super.read(b, off, len);
        consume(res);
        return res;
      }
      private void consume(int minusOne) {
        if (!consumed && minusOne==-1) {
          try {
            EntityUtils.consume(entity);
          } catch (Exception e) {
            // ignored on purpose
          }
          consumed = true;
        }
      }
    };
  }
  
  /**
   * Returns true iff this instance was {@link #close() closed}, otherwise
   * returns false. Note that if you override {@link #close()}, you must call
   * {@code super.close()}, in order for this instance to be properly closed.
   */
  protected final boolean isClosed() {
    return closed;
  }
  
  /**
   * Same as {@link #doAction(HttpResponse, boolean, Callable)} but always do consume at the end.
   */
  protected <T> T doAction(HttpResponse response, Callable<T> call) throws IOException {
    return doAction(response, true, call);
  }
  
  /**
   * Do a specific action and validate after the action that the status is still OK, 
   * and if not, attempt to extract the actual server side exception. Optionally
   * release the response at exit, depending on <code>consume</code> parameter.
   */
  protected <T> T doAction(HttpResponse response, boolean consume, Callable<T> call) throws IOException {
    IOException error = null;
    try {
      return call.call();
    } catch (IOException e) {
      error = e;
    } catch (Exception e) {
      error = new IOException(e);
    } finally {
      try {
        verifyStatus(response);
      } finally {
        if (consume) {
          try {
            EntityUtils.consume(response.getEntity());
          } catch (Exception e) {
            // ignoring on purpose
          }
        }
      }
    }
    throw error; // should not get here
  }
  
  @Override
  public void close() throws IOException {
    closed = true;
  }
  
}
