Java 类org.apache.http.nio.entity.NStringEntity 实例源码

private void bulkIndexMember(List<?> memList) throws Exception {
    StringBuilder buf = new StringBuilder(1024);
    for (Object mem : memList) {
        buf.append("{\"index\": {}}");
        buf.append("\n");
        buf.append(Gson.toJson(mem));
        buf.append("\n");
    }

    long startTime = System.currentTimeMillis();
    RestClient client = Plugin.client;

    HttpEntity entity = new NStringEntity(buf.toString(), ContentType.APPLICATION_JSON);

    Response indexResponse = client.performRequest("POST", "/weike/member/_bulk",
            Collections.<String, String>emptyMap(), entity);

    if (LOG.isDebugEnabled()) {
        LOG.debug("indexMember {}ms", System.currentTimeMillis() - startTime);
        LOG.debug("indexResponse {}", indexResponse.toString());
    }
}

@Override
public boolean bulkIndex(List<String> bulkData) {
    StringBuilder bulkRequestBody = new StringBuilder();
    for (String bulkItem : bulkData) {
        bulkRequestBody.append(actionMetaData);
        bulkRequestBody.append(bulkItem);
        bulkRequestBody.append("\n");
    }
    HttpEntity entity = new NStringEntity(bulkRequestBody.toString(), ContentType.APPLICATION_JSON);
    try {
        Response response = client.performRequest("POST", "/geonames/type/_noop_bulk", Collections.emptyMap(), entity);
        return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK;
    } catch (Exception e) {
        throw new ElasticsearchException(e);
    }
}

/**
 * Perform the replication. All logic is covered in {@link ElasticSearchIndexContentBuilder} so we only need to transmit the JSON to ElasticSearch
 *
 * @param ctx
 * @param tx
 * @param restClient
 * @return
 * @throws ReplicationException
 */
private ReplicationResult doActivate(TransportContext ctx, ReplicationTransaction tx, RestClient restClient) throws ReplicationException, JSONException, IOException {
  ReplicationLog log = tx.getLog();
  ObjectMapper mapper = new ObjectMapper();
  IndexEntry content = mapper.readValue(tx.getContent().getInputStream(), IndexEntry.class);
  if (content != null) {
    log.info(getClass().getSimpleName() + ": Indexing " + content.getPath());
    String contentString = mapper.writeValueAsString(content.getContent());
    log.debug(getClass().getSimpleName() + ": Index-Content: " + contentString);
    LOG.debug("Index-Content: " + contentString);

    HttpEntity entity = new NStringEntity(contentString, "UTF-8");
    Response indexResponse = restClient.performRequest(
            "PUT",
            "/" + content.getIndex() + "/" + content.getType() + "/" + DigestUtils.md5Hex(content.getPath()),
            Collections.<String, String>emptyMap(),
            entity);
    LOG.debug(indexResponse.toString());
    log.info(getClass().getSimpleName() + ": " + indexResponse.getStatusLine().getStatusCode() + ": " + indexResponse.getStatusLine().getReasonPhrase());
    if (indexResponse.getStatusLine().getStatusCode() == HttpStatus.SC_CREATED || indexResponse.getStatusLine().getStatusCode() == HttpStatus.SC_OK) {
      return ReplicationResult.OK;
    }
  }
  LOG.error("Could not replicate");
  return new ReplicationResult(false, 0, "Replication failed");
}

/**
 * Retrieve the informatoin of a realm
 * @param bot The bot instance.
 * @param region The region the realm is in
 * @param realm The realm name
 * @return A Json string containing information about the realm. Returns null if no realm is found.
 */
public static String getRealmInformation(LegendaryBot bot, String region, String realm) {
    HttpEntity entity = new NStringEntity("{ \"query\": { \"match\" : { \"name\" : \""+realm+"\" } } }", ContentType.APPLICATION_JSON);
    try {
        Response response = bot.getElasticSearch().performRequest("POST", "/wow/realm_"+region.toLowerCase()+"/_search", Collections.emptyMap(), entity);
        String jsonResponse = EntityUtils.toString(response.getEntity());
        JSONParser jsonParser = new JSONParser();
        JSONObject obj = (JSONObject) jsonParser.parse(jsonResponse);
        JSONArray hit = (JSONArray) ((JSONObject)obj.get("hits")).get("hits");
        if (hit.size() == 0) {
            return null;
        }
        JSONObject firstItem = (JSONObject) hit.get(0);
        JSONObject source = (JSONObject)  firstItem.get("_source");
        return source.toJSONString();
    } catch (IOException | ParseException e) {
        e.printStackTrace();
    }
    return null;
}

public String search(String q) throws IOException {
    HttpEntity entity1 = new NStringEntity(
            "{\n" +
                    "    \"query\" : {\n" +
                    "    \"match\": { \"dc\":\""+q+"\"} \n" +
                    "}, \n"+
                    "    \"sort\" : [\n" +
                    "    {\"title.keyword\": { \"order\":\"desc\"}} \n" +
                    "], \n"+
                    "\"_source\":\"title\""+
                    "}"
            , ContentType.APPLICATION_JSON);

    Response response = restClient.performRequest("GET", "/harvester/_search", Collections.singletonMap("pretty", "true"),
            entity1);
    String result =  ( EntityUtils.toString(response.getEntity()));
    System.out.println(result);

    return "<pre>"+result+"</pre>"; //pre tag for json, otherwise it didnt show pretty in browser

}

@Override
public boolean advance() throws IOException {
  if (batchIterator.hasNext()) {
    current = batchIterator.next();
    return true;
  } else {
    String requestBody =
        String.format(
            "{\"scroll\" : \"%s\",\"scroll_id\" : \"%s\"}",
            source.spec.getScrollKeepalive(), scrollId);
    HttpEntity scrollEntity = new NStringEntity(requestBody, ContentType.APPLICATION_JSON);
    Response response =
        restClient.performRequest(
            "GET",
            "/_search/scroll",
            Collections.<String, String>emptyMap(),
            scrollEntity);
    JsonNode searchResult = parseResponse(response);
    updateScrollId(searchResult);
    return readNextBatchAndReturnFirstDocument(searchResult);
  }
}

@Override
public void close() throws IOException {
  // remove the scroll
  String requestBody = String.format("{\"scroll_id\" : [\"%s\"]}", scrollId);
  HttpEntity entity = new NStringEntity(requestBody, ContentType.APPLICATION_JSON);
  try {
    restClient.performRequest(
        "DELETE",
        "/_search/scroll",
        Collections.<String, String>emptyMap(),
        entity);
  } finally {
    if (restClient != null) {
      restClient.close();
    }
  }
}

private void flushBatch() throws IOException {
  if (batch.isEmpty()) {
    return;
  }
  StringBuilder bulkRequest = new StringBuilder();
  for (String json : batch) {
    bulkRequest.append(json);
  }
  batch.clear();
  currentBatchSizeBytes = 0;
  Response response;
  String endPoint =
      String.format(
          "/%s/%s/_bulk",
          spec.getConnectionConfiguration().getIndex(),
          spec.getConnectionConfiguration().getType());
  HttpEntity requestBody =
      new NStringEntity(bulkRequest.toString(), ContentType.APPLICATION_JSON);
  response =
      restClient.performRequest(
          "POST",
          endPoint,
          Collections.<String, String>emptyMap(),
          requestBody);
  checkForErrors(response, backendVersion);
}

/** Inserts the given number of test documents into Elasticsearch. */
static void insertTestDocuments(ConnectionConfiguration connectionConfiguration,
    long numDocs, RestClient restClient) throws IOException {
  List<String> data =
      ElasticSearchIOTestUtils.createDocuments(
          numDocs, ElasticSearchIOTestUtils.InjectionMode.DO_NOT_INJECT_INVALID_DOCS);
  StringBuilder bulkRequest = new StringBuilder();
  int i = 0;
  for (String document : data) {
    bulkRequest.append(String.format(
        "{ \"index\" : { \"_index\" : \"%s\", \"_type\" : \"%s\", \"_id\" : \"%s\" } }%n%s%n",
        connectionConfiguration.getIndex(), connectionConfiguration.getType(), i++, document));
  }
  String endPoint = String.format("/%s/%s/_bulk", connectionConfiguration.getIndex(),
      connectionConfiguration.getType());
  HttpEntity requestBody =
      new NStringEntity(bulkRequest.toString(), ContentType.APPLICATION_JSON);
  Response response = restClient.performRequest("POST", endPoint,
      Collections.singletonMap("refresh", "true"), requestBody);
  ElasticsearchIO
      .checkForErrors(response, ElasticsearchIO.getBackendVersion(connectionConfiguration));
}

public void failed(final Exception ex) {
    synchronized (this.httpExchange) {
        if (this.completed) {
            return;
        }
        this.completed = true;
        this.httpExchange.setException(ex);
        HttpAsyncExchange responseTrigger = this.httpExchange.getResponseTrigger();
        if (responseTrigger != null && !responseTrigger.isCompleted()) {
            System.out.println("[client<-proxy] " + this.httpExchange.getId() + " " + ex);
            ConsoleFactory.printToConsole("[client<-proxy] " + this.httpExchange.getId() + " " + ex,true);
            int status = HttpStatus.SC_INTERNAL_SERVER_ERROR;
            HttpResponse response = new BasicHttpResponse(HttpVersion.HTTP_1_0, status,
                    EnglishReasonPhraseCatalog.INSTANCE.getReason(status, Locale.US));
            String message = ex.getMessage();
            if (message == null) {
                message = "Unexpected error";
            }
            response.setEntity(new NStringEntity(message, ContentType.DEFAULT_TEXT));
            responseTrigger.submitResponse(new BasicAsyncResponseProducer(response));
        }
    }
}

/**
 * 
 * @param targetClass
 * @param path
 * @param params
 * @return
 */
public <T>Result<T> execute(Class<T> targetClass,
        String path, Object params) {
    HttpPost post = new HttpPost(getContext().getApiUrl(path));
    String data = convertParamsToJson(params);
    NStringEntity entity = new NStringEntity(data, ContentType.APPLICATION_JSON);
    post.setEntity(entity);

    Result<T> result = null;

    try (CloseableHttpResponse response = httpClient.execute(post)) {
        InputStream in = response.getEntity().getContent();
        if (response.getStatusLine().getStatusCode() == 200) {
            result = new Result<>(context.getObjectMapper()
                    .readValue(in, targetClass));
        } else {
            result = new Result<>(context.getObjectMapper()
                    .readValue(in, ErrorInfo.class));
        }
    } catch (IOException e) {
        throw new IORuntimeException(e);
    }

    return result;
}

private void indexMember(String sellerId, Object mem) throws IOException {
    if (sellerId == null)
        sellerId = "";

    long startTime = System.currentTimeMillis();

    RestClient client = Plugin.client;
    String json = Gson.toJson(mem);

    HttpEntity entity = new NStringEntity(json, ContentType.APPLICATION_JSON);
    String path = "/weike/member";
    if (!"".equals(sellerId)) {
        path += "?routing=" + sellerId;
    }
    Response indexResponse = client.performRequest("POST", path, Collections.<String, String>emptyMap(), entity);

    if (LOG.isDebugEnabled()) {
        LOG.debug("indexMember {}ms", System.currentTimeMillis() - startTime);
        LOG.debug("indexResponse {}", indexResponse.toString());
    }
}

@Test
@Ignore
public void testSearch() throws Exception {
    // JsonObject json = new JsonObject();
    // json.addProperty("from", "0");
    // json.addProperty("size", "10");
    // json.addProperty("explain", true);
    // JsonObject query = new JsonObject();
    // query.add
    // json.add("query", query);
    String json = "{\"explain\":false,\"from\":0,\"size\":1,\"query\":{\"range\":{\"tradeAmount\":{\"gte\":10,\"lte\":2000}}}}";

    long startTime = System.currentTimeMillis();
    HttpEntity entity = new NStringEntity(json, ContentType.APPLICATION_JSON);
    Response response = client.performRequest("GET", "/weike/member/_search", Collections.singletonMap("pretty", "true"), entity);
    LOG.info("search-{} {}ms", EntityUtils.toString(response.getEntity()), System.currentTimeMillis() - startTime);
}

/**
 * Creates a {@link EntityRequest} that can be used to authenticate the user against the server.
 * @param store The {@link Store} this request should be executed against.
 * @return A request that can be used to authenticate the user.
 */
public EntityRequest createLoginRequest(final Store store) {
    final EntityRequest request = new EntityRequest(
            Method.POST,
            this.getUrl(store, URL_AUTH_LOGIN));

    final JsonObject credentials = new JsonObject();
    credentials.addProperty("userName", store.getUsername());
    credentials.addProperty("password", store.getPassword());

    final NStringEntity entity = new NStringEntity(credentials.toString(), CHARSET);
    request.setEntity(entity);

    request.setHeader(Headers.CONTENT_TYPE, APPLICATION_JSON);
    return request;
}

public String doPost(String url, String data, String charset) {
    if (StringUtils.isBlank(url)) {
        return null;
    }
    log.info(" post url=" + url);
    try {
        HttpPost httpPost = new HttpPost(url);
        httpPost.setEntity(new NStringEntity(data, charset));
        CloseableHttpResponse response = httpClient.execute(httpPost);
        int statusCode = response.getStatusLine().getStatusCode();
        if (statusCode != 200) {
            httpPost.abort();
            throw new RuntimeException("HttpClient,error status code :" + statusCode);
        }
        HttpEntity entity = response.getEntity();
        String result = null;
        if (entity != null) {
            result = EntityUtils.toString(entity, charset);
        }
        EntityUtils.consume(entity);
        response.close();
        return result;
    } catch (Exception e) {
        log.error("to request addr=" + url + ", " + e.getMessage());
        e.printStackTrace();
    }
    return null;
}

@Override
public boolean search(String source) {
    HttpEntity searchBody = new NStringEntity(source, StandardCharsets.UTF_8);
    try {
        Response response = client.performRequest("GET", endpoint, Collections.emptyMap(), searchBody);
        return response.getStatusLine().getStatusCode() == HttpStatus.SC_OK;
    } catch (IOException e) {
        throw new ElasticsearchException(e);
    }
}

/**
 * Creates an index and adds an alias to it.
 * @param aliasName
 * @param version
 * @param indexConf must be a valid ES config JSON.
 */
public void createIndex(String aliasName, int version, String indexConf) {
    HttpEntity entity = new NStringEntity(indexConf, ContentType.APPLICATION_JSON);
    String indexName = aliasName + INDEX_VERSION_INFIX + version;
    try {
        restClient.performRequest("PUT", indexName, Collections.emptyMap(), entity);
        addAlias(indexName, aliasName);
    } catch (IOException e) {
        e.printStackTrace();
    }
}

@Override
public void send(String body) throws IOException {
    HttpEntity entity = new NStringEntity(body.toString(), ContentType.APPLICATION_JSON);
    Response response = this.restClient.performRequest(HTTP_METHOD, "_bulk", Collections.emptyMap(), entity);
    if (response.getStatusLine().getStatusCode() >= 300) {
        throw new HttpResponseException(response.getStatusLine().getStatusCode(), response.getStatusLine().getReasonPhrase());
    }
}

public <T> T post(String resource, String body, Class<T> responseClass, boolean assertSuccess)
        throws IOException {
    final Response response = client.performRequest("POST", resource, Collections.singletonMap("pretty", "true"),
            new NStringEntity(body, ContentType.APPLICATION_JSON));
    if (assertSuccess) {
        assertSuccess(response);
    }
    return mapper.readValue(IOUtils.toString(response.getEntity().getContent(), "UTF-8"), responseClass);
}

public <T> T put(String resource, String body ,Class<T> responseClass, boolean assertSuccess) throws IOException {
    final HttpEntity entity = new NStringEntity(body, ContentType.APPLICATION_JSON);
    final Response response = client.performRequest("PUT", resource, Collections.<String, String>emptyMap(),
            entity);
    if (assertSuccess) {
        assertSuccess(response);
    }
    return mapper.readValue(IOUtils.toString(response.getEntity().getContent(), "UTF-8"), responseClass);
}

/**
 * Makes an HTTP POST request using the given URI and optional body.
 *
 * @param uri the URI to use for the POST call
 * @param json Optional, can be null. the JSON to POST
 * @return the resulting HttpFuture instance
 */
public HttpFuture doPost( URI uri, JsonNode json ) {
    HttpPost httpPost = new HttpPost( uri );

    String body = null;
    if( json != null ) {
        body = jsonHelper.serialize( json );
        httpPost.setEntity( new NStringEntity( body, ContentType.create( "application/json", "UTF-8" ) ) );
    }

    return new HttpFuture( this, httpPost, asyncClient.execute( httpPost, null ), body );
}

/**
 * Makes an HTTP PUT request using the given URI and optional body.
 *
 * @param uri the URI to use for the PUT call
 * @param json Optional, can be null. the JSON to PUT
 * @return the resulting HttpFuture instance
 */
public HttpFuture doPut( URI uri, JsonNode json ) {
    HttpPut httpPut = new HttpPut( uri );

    String body = null;
    if( json != null ) {
        body = jsonHelper.serialize( json );
        httpPut.setEntity( new NStringEntity( body, ContentType.create( "application/json", "UTF-8" ) ) );
    }

    return new HttpFuture( this, httpPut, asyncClient.execute( httpPut, null ), body );
}

public Response indexTweet(int id, String user, String message) throws IOException {
    HttpEntity entity = new NStringEntity(
        String.format("{\"user\" : \"%s\", \"message\" : \"%s\"}", user, message),
        ContentType.APPLICATION_JSON);
    Response response = restClient.performRequest(
        "PUT",
        "/twitter/tweet/" + id,
        Collections.<String, String>emptyMap(),
        entity);
    return response;
}

public Response performRequest(String method, String endpoint, Map<String, String> params,
                               JSONObject entity, Header... headers) {
    try {
        HttpEntity payload = new NStringEntity(JsonUtils.convertToString(entity), ContentType.APPLICATION_JSON);
        return restClient.performRequest(method, endpoint, params, payload, HttpAsyncResponseConsumerFactory.DEFAULT, headers);
    } catch (IOException e) {
        throw new RequestException(
                String.format("request exception. method: %s, endpoint: %s, params: %s, entity: %s",
                        method, endpoint, params, entity),
                e);
    }
}

void index(String key, String value) {
  try {
    final HttpEntity entity = new NStringEntity(value, ContentType.APPLICATION_JSON);
    final String endpoint = String.format("%s/%s", "/tweets/tweet", key);
    final Response response = restClient.performRequest("PUT", endpoint, Collections.<String, String>emptyMap(), entity);
    if (response.getStatusLine().getStatusCode() != 200
        && response.getStatusLine().getStatusCode() != 201) {
      throw new IllegalStateException(response.getStatusLine().getReasonPhrase());
    }
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}

public static void main(String[] args) {

        try {
            RestClient client = null;
            String url = "localhost";
            int port = 9200;

            CredentialsProvider credentialsProvider = setCredentials("elastic", "changeme");

            RestClient restClient = createRestClient(url, port, credentialsProvider);

            System.out.println("Connected");

            //index a document
            HttpEntity entity = new NStringEntity(
                    "{\n"
                    + "    \"user\" : \"kimchy\",\n"
                    + "    \"post_date\" : \"2009-11-15T14:12:12\",\n"
                    + "    \"message\" : \"trying out Elasticsearch\"\n"
                    + "}", ContentType.APPLICATION_JSON);

            Response indexResponse = restClient.performRequest(
                    "PUT",
                    "/mds/test/12345678",
                    Collections.<String, String>emptyMap(),
                    entity);

            System.out.println(indexResponse.toString());

        } catch (IOException ex) {
            Logger.getLogger(RestClientTest.class.getName()).log(Level.SEVERE, null, ex);
        }
    }

@Override
public boolean start() throws IOException {
  restClient = source.spec.getConnectionConfiguration().createClient();

  String query = source.spec.getQuery();
  if (query == null) {
    query = "{\"query\": { \"match_all\": {} }}";
  }
  if (source.backendVersion == 5){
    //if there is more than one slice
    if (source.numSlices != null && source.numSlices > 1){
      // add slice to the user query
      String sliceQuery = String
          .format("\"slice\": {\"id\": %s,\"max\": %s}", source.sliceId,
              source.numSlices);
      query = query.replaceFirst("\\{", "{" + sliceQuery + ",");
    }
  }
  Response response;
  String endPoint =
      String.format(
          "/%s/%s/_search",
          source.spec.getConnectionConfiguration().getIndex(),
          source.spec.getConnectionConfiguration().getType());
  Map<String, String> params = new HashMap<>();
  params.put("scroll", source.spec.getScrollKeepalive());
  if (source.backendVersion == 2){
    params.put("size", String.valueOf(source.spec.getBatchSize()));
    if (source.shardPreference != null) {
      params.put("preference", "_shards:" + source.shardPreference);
    }
  }
  HttpEntity queryEntity = new NStringEntity(query,
      ContentType.APPLICATION_JSON);
  response =
      restClient.performRequest("GET", endPoint, params, queryEntity);
  JsonNode searchResult = parseResponse(response);
  updateScrollId(searchResult);
  return readNextBatchAndReturnFirstDocument(searchResult);
}

void testWrite() throws Exception {
  List<String> data =
      ElasticSearchIOTestUtils.createDocuments(
          numDocs, ElasticSearchIOTestUtils.InjectionMode.DO_NOT_INJECT_INVALID_DOCS);
    pipeline
      .apply(Create.of(data))
      .apply(ElasticsearchIO.write().withConnectionConfiguration(connectionConfiguration));
  pipeline.run();

  long currentNumDocs = ElasticSearchIOTestUtils
      .refreshIndexAndGetCurrentNumDocs(connectionConfiguration, restClient);
  assertEquals(numDocs, currentNumDocs);

  String requestBody =
      "{\n"
      + "  \"query\" : {\"match\": {\n"
      + "    \"scientist\": \"Einstein\"\n"
      + "  }}\n"
      + "}\n";
  String endPoint = String.format("/%s/%s/_search", connectionConfiguration.getIndex(),
      connectionConfiguration.getType());
  HttpEntity httpEntity = new NStringEntity(requestBody, ContentType.APPLICATION_JSON);
  Response response =
      restClient.performRequest(
          "GET",
          endPoint,
          Collections.<String, String>emptyMap(),
          httpEntity);
  JsonNode searchResult = parseResponse(response);
  int count = searchResult.path("hits").path("total").asInt();
  assertEquals(numDocs / NUM_SCIENTISTS, count);
}

public ResponseHits getResponseHits(final String method, final String url, final String requestBody, final TypeReference typeReference) {
    ResponseHits responseHits = null;
    RestClient restClient = this.elasticSearchClient.getRestClient();
    String endpoint = "/" + this.elasticSearchConfiguration.getIndex() + url;
    try {
        HttpEntity entity = new NStringEntity(requestBody, ContentType.APPLICATION_JSON);
        Response response = restClient.performRequest(method, endpoint, Collections.emptyMap(), entity);
        responseHits = new ObjectMapper().readValue(response.getEntity().getContent(), typeReference);
    } catch (IOException e) {
        e.printStackTrace(); //TODO Gérer l'exception
        throw new RuntimeException(e);
    }
    return responseHits;
}

protected void emit(final Map event) {
    String _index = (String) this.indexRender.render(event);
    String _indexType = (String) indexTypeRender.render(event);
    String requestBody;
    Response response = null;

    addActionList(event, _index, _indexType);
    if (this.actionList.size() / 2 >= this.bulkActions) {
        try {

            requestBody = actionList.stream().map(JSONValue::toJSONString).collect(Collectors.joining("\n")) + "\n";
            log.info(requestBody);
            response = restClient.performRequest(
                    "POST",
                    BULKPATH,
                    Collections.<String, String>emptyMap(),
                    new NStringEntity(
                            requestBody,
                            ContentType.APPLICATION_JSON
                    )
            );
            log.info(response.toString());
        } catch (IOException e) {
            log.error("Bulk index es Error:", e);
            if (response != null)
                log.error("Response Code is " + response.getStatusLine().toString());
        } finally {
            actionList.clear();
        }
    }
}

protected void writeAlert(Action a) throws Exception {
    String url = a.getProperties().get(PROP_URL);
    String index = a.getProperties().get(PROP_INDEX);
    String type = a.getProperties().get(PROP_TYPE);
    String[] urls = url.split(",");
    HttpHost[] hosts = new HttpHost[urls.length];
    for (int i=0; i<urls.length; i++) {
        hosts[i] = HttpHost.create(urls[0]);
    }
    RestClient client = RestClient.builder(hosts)
            .setHttpClientConfigCallback(httpClientBuilder -> {
                httpClientBuilder.useSystemProperties();
                CredentialsProvider credentialsProvider = checkBasicCredentials(a);
                if (credentialsProvider != null) {
                    httpClientBuilder.setDefaultCredentialsProvider(credentialsProvider);
                }
                return httpClientBuilder;
            }).build();

    HttpEntity document = new NStringEntity(transform(a), ContentType.APPLICATION_JSON);
    String endpoint = "/" + index + "/" + type;
    Header[] headers = checkHeaders(a);
    Response response = headers == null ? client.performRequest("POST", endpoint, Collections.EMPTY_MAP, document) :
            client.performRequest("POST", endpoint, Collections.EMPTY_MAP, document, headers);
    log.debugf(response.toString());
    client.close();
}

private static HttpEntity createTextEntity(String text) throws UnsupportedEncodingException
{
    NStringEntity entity = new NStringEntity(MessageFormat.format("<html><body><h1>{0}</h1></body></html>", text), //$NON-NLS-1$
            IOUtil.UTF_8);
    entity.setContentType(HTML_TEXT_TYPE + HTTP.CHARSET_PARAM + IOUtil.UTF_8);
    return entity;
}

@Override
public void handle(final HttpRequest request, final HttpAsyncExchange httpexchange, final HttpContext context) {
    executor.schedule(new Runnable() {
        @Override
        public void run() {
            HttpResponse response = httpexchange.getResponse();
            response.setStatusCode(HttpStatus.SC_OK);
            response.setEntity(new NStringEntity(("Slow hello world"), ContentType.create("text/html", "UTF-8")));
            httpexchange.submitResponse();
        }
    }, 50, TimeUnit.MILLISECONDS);
}

private void sendError(Exception e, HttpAsyncExchange httpexchange) {
    e.printStackTrace();
    HttpResponse response = httpexchange.getResponse();
    response.setStatusCode(500);
    StringWriter stringWriter = new StringWriter();
    e.printStackTrace(new PrintWriter(stringWriter));
    response.setEntity(new NStringEntity(stringWriter.toString(),
            ContentType.create("text/plain", "UTF-8")));
    httpexchange.submitResponse();
}

@Override
public void handle(final HttpRequest request, final HttpAsyncExchange httpexchange, final HttpContext context) {
    HttpResponse response = httpexchange.getResponse();
    response.setStatusCode(HttpStatus.SC_OK);
    response.setEntity(new NStringEntity("Hello world", ContentType.create("text/html", "UTF-8")));
    httpexchange.submitResponse();
}

/**
 * 
 * @param path
 * @param params
 * @param futureCallback
 * @return
 */
public Future<HttpResponse> execute(String path, Object params,
        FutureCallback<HttpResponse> futureCallback) {
    HttpPost post = new HttpPost(getContext().getApiUrl(path));
    String data = convertParamsToJson(params);
    NStringEntity entity = new NStringEntity(data,
            ContentType.APPLICATION_JSON);
    post.setEntity(entity);
    Future<HttpResponse> future = httpAsyncClient.execute(post,
            futureCallback);
    return future;
}

@Test
@Ignore
public void testCount() throws Exception {
    // JsonObject json = new JsonObject();
    // json.addProperty("explain", true);
    // JsonObject query = new JsonObject();
    // JsonObject term = new JsonObject();
    // term.addProperty("sellerId", 807426238);
    // query.add("term", term);
    // json.add("query", query);
    // XContentBuilder json = XContentFactory.jsonBuilder();
    // json.startObject().field("explain", true).field("query",
    // XContentFactory.jsonBuilder().);

    String json = "{\"explain\":true,\"query\":{\"range\":{\"tradeAmount\":{\"gte\":10,\"lte\":500}}}}";
    json = "{\"query\" : {\"term\" : { \"sellerId\" :  897258160}}}";

    long startTime = System.currentTimeMillis();
    HttpEntity entity = new NStringEntity(json.toString(), ContentType.APPLICATION_JSON);
    Response response = client.performRequest("GET", "/weike/member/_count", Collections.singletonMap("pretty", "true"), entity);

    // LOG.info(XContentFactory.jsonBuilder().startObject().field("gender",
    // "male").endObject().string());
    LOG.info("count-{} {}ms", EntityUtils.toString(response.getEntity()), System.currentTimeMillis() - startTime);

    json = "{\"actions\" : ["
            // + "{ \"remove\" : { \"index\" : \"test1\", \"alias\" :
            // \"alias1\" } },"
            + "{ \"add\" : { \"index\" : \"weike\", \"alias\" : \"wkalias1\"} }" + "]}";
    query("POST", json, "/_aliases", "add");
    query("GET", null, "/_aliases/wkalias1", "get");

    json = "{\"query\" : {\"term\" : { \"sellerId\" :  897258160}}}";
    query("GET", json, "/wkalias1/member/_count", "alias count");

    query("DELETE", null, "/_all/_aliases/wkalias1", "delete");
}

@Test
@Ignore
public void testDeleteScoll() throws IOException {
    String json = "{\"explain\":false,\"scroll_id\":[\"\",\"\"]}";

    long startTime = System.currentTimeMillis();
    HttpEntity entity = new NStringEntity(json, ContentType.APPLICATION_JSON);
    Response response = client.performRequest("DELETE", "/weike/member/_search/scroll", Collections.singletonMap("pretty", "true"), entity);
    LOG.info("scroll search-{} {}ms", EntityUtils.toString(response.getEntity()), System.currentTimeMillis() - startTime);
}

@Test
@Ignore
public void testReq() throws Exception {
    // Response response = client.performRequest("GET", "/",
    // Collections.singletonMap("pretty", "true"));
    // System.out.println(EntityUtils.toString(response.getEntity()));

    // index a document
    HttpEntity entity = new NStringEntity("{\"sellerId\":1," + "\"user\" : \"kimchy\"," + "\"post_date\" : \"2009-11-15T14:12:12\","
            + "    \"message\" : \"trying out Elasticsearch\",\"mobile\":\"18616020611\"" + "}", ContentType.APPLICATION_JSON);
    Response indexResponse = client.performRequest("POST", "/weike/member/", Collections.<String, String> emptyMap(), entity);

}

@Test
public void testReq() throws Exception {
    Response response = client.performRequest("GET", "/", Collections.singletonMap("pretty", "true"));
    System.out.println(EntityUtils.toString(response.getEntity()));

    // index a document
    HttpEntity entity = new NStringEntity("{\n" + "    \"user\" : \"kimchy\",\n" + "    \"post_date\" : \"2009-11-15T14:12:12\",\n"
            + "    \"message\" : \"trying out Elasticsearch\"\n" + "}", ContentType.APPLICATION_JSON);

    String u = URLEncoder.encode("root:dzh", "utf-8");
    BasicHeader auth = new BasicHeader("Authorization", "Basic " + u);
    Response indexResponse = client.performRequest("PUT", "/twitter/tweet/1", Collections.<String, String> emptyMap(), entity);

}