/*
    * 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 is a list of 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 if error writing
    * @throws InterruptedException if shutdown has been initiated
    */
   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 is HTTP method
    * @return the list of commited status
    * @throws InterruptedException if shutdown has been initiated
    * @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() {
   }
 }