import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
public PercolateResponse percolate(String[] indexNames, String documentType, Object tuple)
{
  XContentBuilder docBuilder;
  try {

    docBuilder = XContentFactory.jsonBuilder().startObject();
    docBuilder.field("doc").startObject(); //This is needed to designate the document
    docBuilder.field("content", tuple);
    docBuilder.endObject(); //End of the doc field
    docBuilder.endObject();//End of the JSON root object

    return client.preparePercolate().setIndices(indexNames)
        .setDocumentType(documentType)
        .setSource(docBuilder)
        .execute().actionGet();
  } catch (IOException e) {
    DTThrowable.rethrow(e);
  }
  return null;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
ArrayList<String> fetchMatchingAlertTerms(AbstractPost post){
    ArrayList<String> matchingTerms = new ArrayList<String>();
    try {
        XContentBuilder docBuilder = XContentFactory.jsonBuilder().startObject();
        docBuilder.field("doc").startObject(); //This is needed to designate the document
        docBuilder.field(ALERT_FIELD, post.text);
        docBuilder.endObject(); //End of the doc field
        docBuilder.endObject(); //End of the JSON root object
        //Percolate
        PercolateResponse response = ElasticsearchConnectionManager.getClient().preparePercolate()
                .setIndices(ALERT_INDEX)
                .setDocumentType(post.getType())
                .setSource(docBuilder).execute().actionGet();
        //Iterate over the results and add the registered alert term
        for(PercolateResponse.Match match : response) {
            matchingTerms.add(match.getId().toString());
        }
    } catch (IOException e) {
        System.out.println("Error! Can't check if a post is matching a percolator alert");
        e.printStackTrace();
    }
    return matchingTerms;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
@Override
public void process(Object tuple)
{

  PercolateResponse response = store.percolate(new String[] {indexName}, documentType, tuple);
  outputPort.emit(response);
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
@Override
public List<PercolateQuery> percolate(List<Document> docs) {
    final Stopwatch watch = LoggingUtils.createStopWatch();
    final List<PercolateQuery> results = new ArrayList<>();
    final MultiPercolateRequestBuilder multiPercolate = client.prepareMultiPercolate();

    for (final Document doc : docs) {
        if (doc.getPercolate() != null) {
            multiPercolate.add(
                    new PercolateRequestBuilder(client).setDocumentType(doc.get_type())
                            .setSize(doc.getPercolate().getMaxMatches())
                            .setPercolateDoc(new PercolateSourceBuilder.DocBuilder().setDoc(doc.get_jsonObject())));
        }
    }

    final MultiPercolateResponse.Item[] responses = multiPercolate.get().getItems();
    for (final MultiPercolateResponse.Item response : responses) {
        if (!response.isFailure()) {
            for (final PercolateResponse.Match match : response.getResponse().getMatches()) {
                results.add(new PercolateQuery().setId(match.getId().toString()));
            }
        }
    }

    LoggingUtils.stopAndLogStopWatch(logger, watch, "Percolate");

    return results;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
@Override
protected XContentBuilder toXContent(PercolateRequest request, PercolateResponse response, XContentBuilder builder) throws IOException {
    builder.startObject();
    builder.field(Fields.OK, true);
    builder.startArray(Fields.MATCHES);
    for (String match : response) {
        builder.value(match);
    }
    builder.endArray();
    builder.endObject();
    return builder;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
private void testPercolate() throws IOException {

        // This is the query we're registering in the percolator
        QueryBuilder qb = termQuery("content", "amazing");

        // Index the query = register it in the percolator
        client.prepareIndex("_percolator", "myIndexName", "myDesignatedQueryName") //
                .setSource(qb.buildAsBytes()) //
                .setRefresh(true) // Needed when the query shall be available
                                  // immediately
                .execute().actionGet();

        // Build a document to check against the percolator
        XContentBuilder docBuilder = jsonBuilder().startObject();
        docBuilder.field("doc").startObject(); // This is needed to designate
                                               // the document
        docBuilder.field("content", "This is amazing!");
        docBuilder.endObject(); // End of the doc field
        docBuilder.endObject(); // End of the JSON root object
        // Percolate
        PercolateResponse response = client.preparePercolate("myIndexName", "myDocumentType") //
                .setSource(docBuilder) //
                .execute() //
                .actionGet();
        // Iterate over the results
        for (String result : response) {
            // Handle the result which is the name of
            // the query in the percolator
        }
    }
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
ReduceResult(long count, PercolateResponse.Match[] matches, InternalAggregations reducedAggregations) {
    this.count = count;
    this.matches = matches;
    this.reducedAggregations = reducedAggregations;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
public PercolateResponse.Match[] matches() {
    return matches;
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
void executePercolate(final PercolateRequest percolateRequest, final RestChannel restChannel, final Client client) {
    client.percolate(percolateRequest, new RestToXContentListener<PercolateResponse>(restChannel));
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
/**
 *
 */
private void checkPercolateResponse()
{
  ElasticSearchPercolatorOperator oper = new ElasticSearchPercolatorOperator();
  oper.hostName = HOST_NAME;
  oper.port = PORT;
  oper.indexName = INDEX_NAME;
  oper.documentType = DOCUMENT_TYPE;
  oper.setup(null);

  String[] messages = {"{content:'This will match only with malhar'}",

    "{content:'This will match only with github'}",

    "{content:'This will match with both github and malhar'}",

    "{content:'This will not match with any of them'}"

  };

  String[][] matches = {

    {MALHAR_TOPIC},

    {GITHUB_TOPIC},

    {GITHUB_TOPIC, MALHAR_TOPIC},

    {}

  };

  CollectorTestSink<PercolateResponse> sink = new CollectorTestSink<PercolateResponse>();
  oper.outputPort.setSink((CollectorTestSink)sink);

  for (String message : messages) {
    oper.inputPort.process(message);
  }

  int i = 0;
  for (PercolateResponse response : sink.collectedTuples) {
    List<String> matchIds = new ArrayList<String>();
    for (Match match : response.getMatches()) {
      matchIds.add(match.getId().toString());
    }
    Collections.sort(matchIds);

    Assert.assertArrayEquals(matchIds.toArray(), matches[i]);
    i++;
  }
}
 

import org.elasticsearch.action.percolate.PercolateResponse; //导入依赖的package包/类
@Override
protected ActionFuture<PercolateResponse> doExecute(PercolateRequest request) {
    return client.percolate(request);
}