package com.thinkbiganalytics.esbulkloader;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;

import org.apache.commons.httpclient.HttpClient;
import org.apache.commons.httpclient.HttpStatus;
import org.apache.commons.httpclient.methods.PostMethod;
import org.apache.commons.httpclient.methods.StringRequestEntity;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

/**
 * 
 * 
 */
public class ElasticsearchRecordWriter extends RecordWriter<Text, Text> {
    private static final String DEFAULT_ES_PORT = "9200";
    private static final String DEFAULT_ES_HOST = "localhost";
    private static final String ESBL_PORT = "esbl.port";
    private static final String ESBL_HOST = "esbl.host";

    private static Log log = LogFactory.getLog(ElasticsearchRecordWriter.class);

    PostMethod method;
    final String apiUrl;

    public ElasticsearchRecordWriter(TaskAttemptContext context) {
	String host = context.getConfiguration().get(ESBL_HOST);
	String port = context.getConfiguration().get(ESBL_PORT);
	host = (null == host ? DEFAULT_ES_HOST : host);
	port = (null == port ? DEFAULT_ES_PORT : port);
	apiUrl = String.format("http://%s:%s/_bulk", host, port);
    }

    @Override
    public void write(Text key, Text value) throws IOException,
	    InterruptedException {

	log.info(value.toString());
	method = new PostMethod(apiUrl);
	method.setRequestEntity(new StringRequestEntity(value.toString(),
		"text/json", "UTF-8"));

	BufferedReader br = null;
	HttpClient client = new HttpClient();
	try {
	    int returnCode = client.executeMethod(method);

	    if (returnCode == HttpStatus.SC_NOT_IMPLEMENTED) {
		log.error(String.format(
			"The Post method is not implemented by this URI (%s)",
			apiUrl));
		// still consume the response body
		method.getResponseBodyAsString();
	    } else {
		br = new BufferedReader(new InputStreamReader(
			method.getResponseBodyAsStream()));
		// if (log.isInfoEnabled()) {
		String readLine;
		while (((readLine = br.readLine()) != null)) {
		    System.out.println(readLine);
		}
		// }
	    }
	} catch (Exception e) {
	    log.error(e);
	} finally {
	    if (br != null) {
		try {
		    br.close();
		} catch (Exception fe) {
		    log.error(fe);
		}
	    }
	}
    }

    @Override
    public void close(TaskAttemptContext arg0) throws IOException,
	    InterruptedException {
	method.releaseConnection();
    }

}
@@ -126,24 +126,69 @@
 treamReader;
+%0Aimport java.io.UnsupportedEncodingException;
 %0A%0Aimport org
@@ -712,16 +712,146 @@
 Text%3E %7B%0A
+    private static final int DEFAULT_BUFFER_SIZE = 100000;%0A    private static final String ESBL_BUFFER_SIZE = %22esbl.buffer_size%22;%0A
     priv
@@ -1164,16 +1164,24 @@
 );%0A%0A    
+private 
 PostMeth
@@ -1195,16 +1195,24 @@
 od;%0A    
+private 
 final St
@@ -1223,16 +1223,90 @@
  apiUrl;
+%0A    private final int bufferSize;%0A    private final StringBuffer buffer ;
 %0A%0A    pu
@@ -1482,16 +1482,81 @@
 _PORT);%0A
+%09String size = context.getConfiguration().get(ESBL_BUFFER_SIZE);%0A
 %09host = 
@@ -1645,16 +1645,133 @@
  port);%0A
+%09bufferSize = (null == size ? DEFAULT_BUFFER_SIZE : Integer.parseInt(size));%0A%09buffer = new StringBuffer(bufferSize);%0A
 %09apiUrl 
@@ -1940,35 +1940,473 @@
 %7B%0A%0A%09
-log.info(value.toString());
+buffer.append(value.toString());%0A%09log.info(value.toString());%0A%09%0A%09checkFlush();%0A    %7D%0A%0A    @Override%0A    public void close(TaskAttemptContext arg0) throws IOException,%0A%09    InterruptedException %7B%0A%09flush();%0A%09method.releaseConnection();%0A    %7D%0A%0A    private void checkFlush() throws UnsupportedEncodingException %7B%0A%09if (buffer.length()%3Ethis.bufferSize) %7B%0A%09    flush();%0A%09    buffer.setLength(0);%0A%09%7D%0A    %7D%0A    %0A    private void flush() throws UnsupportedEncodingException %7B
 %0A%09me
@@ -2485,21 +2485,22 @@
 tEntity(
-value
+buffer
 .toStrin
@@ -3319,150 +3319,5 @@
  %7D%0A%0A
-    @Override%0A    public void close(TaskAttemptContext arg0) throws IOException,%0A%09    InterruptedException %7B%0A%09method.releaseConnection();%0A    %7D%0A%0A
 %7D

add flushing logic