/*
 * 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.hadoop.chukwa.datacollection.sender;


import java.io.BufferedReader;
import java.io.ByteArrayInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.io.OutputStream;
import java.nio.charset.Charset;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpException;
import org.apache.commons.httpclient.HttpMethod;
import org.apache.commons.httpclient.HttpMethodBase;
import org.apache.commons.httpclient.HttpMethodRetryHandler;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.MultiThreadedHttpConnectionManager;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.RequestEntity;
import org.apache.commons.httpclient.params.HttpMethodParams;
import org.apache.hadoop.chukwa.Chunk;
import org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor;
import org.apache.hadoop.chukwa.datacollection.sender.metrics.HttpSenderMetrics;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.DataOutputBuffer;
import org.apache.hadoop.io.compress.CompressionCodec;
import org.apache.hadoop.io.compress.CompressionOutputStream;
import org.apache.hadoop.util.ReflectionUtils;
import org.apache.log4j.Logger;

/**
 * Encapsulates all of the http setup and connection details needed for chunks
 * to be delivered to a collector.
 * 
 * This class should encapsulate the details of the low level data formatting.
 * The Connector is responsible for picking what to send and to whom;
 * retry policy is encoded in the collectors iterator.
 * 
 * This class is not thread safe. Synchronization is the caller's responsibility.
 * 
 * <p>
 * On error, tries the list of available collectors, pauses for a minute, and
 * then repeats.
 * </p>
 * <p>
 * Will wait forever for collectors to come up.
 * </p>
 */
public class ChukwaHttpSender implements ChukwaSender {
  final int MAX_RETRIES_PER_COLLECTOR; // fast retries, in http client
  final int SENDER_RETRIES;
  final int WAIT_FOR_COLLECTOR_REBOOT;
  final int COLLECTOR_TIMEOUT;
  
  public static final String COLLECTOR_TIMEOUT_OPT = "chukwaAgent.sender.collectorTimeout";
  // FIXME: this should really correspond to the timer in RetryListOfCollectors

  static final HttpSenderMetrics metrics = new HttpSenderMetrics("chukwaAgent", "httpSender");
  
  static Logger log = Logger.getLogger(ChukwaHttpSender.class);
  static HttpClient client = null;
  static MultiThreadedHttpConnectionManager connectionManager = null;
  String currCollector = null;
  int postID = 0;

  protected Iterator<String> collectors;
  
  boolean COMPRESS;
  String CODEC_NAME;
  CompressionCodec codec;

  static {
    connectionManager = new MultiThreadedHttpConnectionManager();
    client = new HttpClient(connectionManager);
    connectionManager.closeIdleConnections(1000);
  }

  public static class CommitListEntry {
    public Adaptor adaptor;
    public long uuid;
    public long start; //how many bytes of stream
    public CommitListEntry(Adaptor a, long uuid, long start) {
      adaptor = a;
      this.uuid = uuid;
      this.start = start;
    }
  }

  // FIXME: probably we're better off with an EventListRequestEntity
  static class BuffersRequestEntity implements RequestEntity {
    List<DataOutputBuffer> buffers;
    boolean compress;
    CompressionCodec codec;

    public BuffersRequestEntity(List<DataOutputBuffer> buf, boolean compress, CompressionCodec codec) {
      buffers = buf;
      this.compress = compress;
      this.codec = codec;
    }
    
    private long getUncompressedContentLenght(){
        long len = 4;// first we send post length, then buffers
        for (DataOutputBuffer b : buffers)
          len += b.getLength();
        return len;
    }

    public long getContentLength() {
    	if(compress) {
    		return -1;
    	}
    	else {
    		return getUncompressedContentLenght();
    	}
    }

    public String getContentType() {
      return "application/octet-stream";
    }

    public boolean isRepeatable() {
      return true;
    }

    private void doWriteRequest( DataOutputStream out ) throws IOException {
        out.writeInt(buffers.size());
        for (DataOutputBuffer b : buffers)
      	 out.write(b.getData(), 0, b.getLength());
    }
    
    public void writeRequest(OutputStream out) throws IOException {
      if(compress) {
          CompressionOutputStream cos = codec.createOutputStream(out);
          DataOutputStream dos = new DataOutputStream( cos);
          doWriteRequest( dos);
          cos.finish();
      }
      else {
          DataOutputStream dos = new DataOutputStream( out);
          doWriteRequest( dos);
      }
    }
  }

  public ChukwaHttpSender(Configuration c) {
    // setup default collector
    ArrayList<String> tmp = new ArrayList<String>();
    this.collectors = tmp.iterator();

    MAX_RETRIES_PER_COLLECTOR = c.getInt("chukwaAgent.sender.fastRetries", 4);
    SENDER_RETRIES = c.getInt("chukwaAgent.sender.retries", 144000);
    WAIT_FOR_COLLECTOR_REBOOT = c.getInt("chukwaAgent.sender.retryInterval",
        20 * 1000);
    COLLECTOR_TIMEOUT = c.getInt(COLLECTOR_TIMEOUT_OPT, 30*1000);
    COMPRESS = c.getBoolean("chukwaAgent.output.compress", false);
    if( COMPRESS) {
	    CODEC_NAME = c.get( "chukwaAgent.output.compression.type", "org.apache.hadoop.io.compress.DefaultCodec");
	    Class<?> codecClass = null;
	    try {
			codecClass = Class.forName( CODEC_NAME);
			codec = (CompressionCodec) ReflectionUtils.newInstance(codecClass, c);
			log.info("codec " + CODEC_NAME + " loaded for network compression");
		} catch (ClassNotFoundException e) {
			log.warn("failed to create codec " + CODEC_NAME + ". Network compression won't be enabled.", e);
			COMPRESS = false;
		}
    }
  }

  /**
   * Set up a list of connectors for this client to send {@link Chunk}s to
   * 
   * @param collectors
   */
  public void setCollectors(Iterator<String> collectors) {
    this.collectors = collectors;
    // setup a new destination from our list of collectors if one hasn't been
    // set up
    if (currCollector == null) {
      if (collectors.hasNext()) {
        currCollector = collectors.next();
      } else
        log.error("No collectors to try in send(), won't even try to do doPost()");
    }
  }

  /**
   * grab all of the chunks currently in the chunkQueue, stores a copy of them
   * locally, calculates their size, sets them up
   * 
   * @return array of chunk id's which were ACKed by collector
   */
  @Override
  public List<CommitListEntry> send(List<Chunk> toSend)
      throws InterruptedException, IOException {
    List<DataOutputBuffer> serializedEvents = new ArrayList<DataOutputBuffer>();
    List<CommitListEntry> commitResults = new ArrayList<CommitListEntry>();

    int thisPost = postID++;
    int toSendSize = toSend.size();
    log.info("collected " + toSendSize + " chunks for post_"+thisPost);

    // Serialize each chunk in turn into it's own DataOutputBuffer and add that
    // buffer to serializedEvents
    for (Chunk c : toSend) {
      DataOutputBuffer b = new DataOutputBuffer(c.getSerializedSizeEstimate());
      try {
        c.write(b);
      } catch (IOException err) {
        log.error("serialization threw IOException", err);
      }
      serializedEvents.add(b);
      // store a CLE for this chunk which we will use to ack this chunk to the
      // caller of send()
      // (e.g. the agent will use the list of CLE's for checkpointing)
      log.info("chunk seqID:"+c.getSeqID());
      commitResults.add(new CommitListEntry(c.getInitiator(), c.getSeqID(), 
         c.getSeqID() - c.getData().length));
    }
    toSend.clear();

    // collect all serialized chunks into a single buffer to send
    RequestEntity postData = new BuffersRequestEntity(serializedEvents, COMPRESS, codec);

    PostMethod method = new PostMethod();
    method.setRequestEntity(postData);
    StringBuilder sb = new StringBuilder( ">>>>>> HTTP post_");
    sb.append( thisPost).append( " to ").append( currCollector).append( " length = ");
	if( COMPRESS) {
		sb.append( ((BuffersRequestEntity)postData).getUncompressedContentLenght())
			.append( " of uncompressed data");
	}
	else {
		sb.append( postData.getContentLength());
	}
	log.info( sb);

    List<CommitListEntry> results =  postAndParseResponse(method, commitResults);
    log.info("post_" + thisPost + " sent " + toSendSize + " chunks, got back " + results.size() + " acks");
    return results;
  }
  
  /**
   * 
   * @param method the data to push
   * @param expectedCommitResults the list
   * @return the list of committed chunks
   * @throws IOException
   * @throws InterruptedException
   */
  public List<CommitListEntry> postAndParseResponse(PostMethod method, 
        List<CommitListEntry> expectedCommitResults)
  throws IOException, InterruptedException{
    reliablySend(method, "chukwa"); //FIXME: shouldn't need to hardcode this here
    return expectedCommitResults;
  }

  /**
   *  Responsible for executing the supplied method on at least one collector
   * @param method
   * @return
   * @throws InterruptedException
   * @throws IOException if no collector responds with an OK
   */
  protected List<String> reliablySend(HttpMethodBase method, String pathSuffix) throws InterruptedException, IOException {
    int retries = SENDER_RETRIES;
    while (currCollector != null) {
      // need to pick a destination here
      try {

        // send it across the network    
        List<String> responses = doRequest(method, currCollector+ pathSuffix);

        retries = SENDER_RETRIES; // reset count on success

        return responses;
      } catch (Throwable e) {
        log.error("Http post exception on "+ currCollector +": "+ e.toString());
        log.debug("Http post exception on "+ currCollector, e);
        ChukwaHttpSender.metrics.httpThrowable.inc();
        if (collectors.hasNext()) {
          ChukwaHttpSender.metrics.collectorRollover.inc();
          boolean repeatPost = failedCollector(currCollector);
          currCollector = collectors.next();
          if(repeatPost)
            log.info("Found a new collector to roll over to, retrying HTTP Post to collector "
                + currCollector);
          else {
            log.info("Using " + currCollector + " in the future, but not retrying this post");
            break;
          }
        } else {
          if (retries > 0) {
            log.warn("No more collectors to try rolling over to; waiting "
                + WAIT_FOR_COLLECTOR_REBOOT + " ms (" + retries
                + " retries left)");
            Thread.sleep(WAIT_FOR_COLLECTOR_REBOOT);
            retries--;
          } else {
            log.error("No more collectors to try rolling over to; aborting post");
            throw new IOException("no collectors");
          }
        }
      } finally {
        // be sure the connection is released back to the connection manager
        method.releaseConnection();
      }
    } // end retry loop
    return new ArrayList<String>();
  }

  /**
   * A hook for taking action when a collector is declared failed.
   * Returns whether to retry current post, or junk it
   * @param downCollector
   */
  protected boolean failedCollector(String downCollector) {
    log.debug("declaring "+ downCollector + " down");
    return true;
  }

  /**
   * Responsible for performing a single operation to a specified collector URL.
   * 
   * @param dest the URL being requested. (Including hostname)
   */
  protected List<String> doRequest(HttpMethodBase method, String dest)
      throws IOException, HttpException {

    HttpMethodParams pars = method.getParams();
    pars.setParameter(HttpMethodParams.RETRY_HANDLER,
        (Object) new HttpMethodRetryHandler() {
          public boolean retryMethod(HttpMethod m, IOException e, int exec) {
            return !(e instanceof java.net.ConnectException)
                && (exec < MAX_RETRIES_PER_COLLECTOR);
          }
        });

    pars.setParameter(HttpMethodParams.SO_TIMEOUT, Integer.valueOf(COLLECTOR_TIMEOUT));

    method.setParams(pars);
    method.setPath(dest);

    // Send POST request
    ChukwaHttpSender.metrics.httpPost.inc();
    
    int statusCode = client.executeMethod(method);

    if (statusCode != HttpStatus.SC_OK) {
      ChukwaHttpSender.metrics.httpException.inc();
      
      if (statusCode == HttpStatus.SC_REQUEST_TIMEOUT ) {
        ChukwaHttpSender.metrics.httpTimeOutException.inc();
      }
      
      log.error(">>>>>> HTTP response from " + dest + " statusLine: " + method.getStatusLine());
      // do something aggressive here
      throw new HttpException("got back a failure from server");
    }
    // implicitly "else"
    log.info(">>>>>> HTTP Got success back from "+ dest + "; response length "
            + method.getResponseContentLength());

    // FIXME: should parse acks here
    InputStream rstream = null;

    // Get the response body
    byte[] resp_buf = method.getResponseBody();
    rstream = new ByteArrayInputStream(resp_buf);
    BufferedReader br = new BufferedReader(new InputStreamReader(rstream, Charset.forName("UTF-8")));
    String line;
    List<String> resp = new ArrayList<String>();
    while ((line = br.readLine()) != null) {
      if (log.isDebugEnabled()) {
        log.debug("response: " + line);
      }
      resp.add(line);
    }
    return resp;
  }

  @Override
  public void stop() {
  }
}