private Double findAggregationValue(MetricContract metricContract, SearchResponse searchResponse) {
    String aggType = metricContract.getAggregation_type();
    if (aggType.equalsIgnoreCase("max")) {
        Max max = searchResponse.getAggregations().get("maxNumber");
        return max.getValue();
    }
    if (aggType.equalsIgnoreCase("min")) {
        Min min = searchResponse.getAggregations().get("minNumber");
        return min.getValue();
    }
    if (aggType.equalsIgnoreCase("avg")) {
        Avg avg = searchResponse.getAggregations().get("avgNumber");
        return avg.getValue();
    }
    if (aggType.equalsIgnoreCase("sum")) {
        Sum sum = searchResponse.getAggregations().get("sumNumber");
        return sum.getValue();
    }

    throw new IllegalArgumentException("unsupported aggregation type: " + aggType);
}
 

private static <A extends Aggregation> void route(Bucket bucket, A a) {
    if (a instanceof Terms) {
        bucket.setBuckets(terms(a));
    } else if (a instanceof Range) {
        bucket.setBuckets(range(a));
    } else if (a instanceof Histogram) {
        bucket.setBuckets(range(a));
    } else if (a instanceof Avg) {
        bucket.setAvg(avg(a));
    } else if (a instanceof Min) {
        bucket.setMin(min(a));
    } else if (a instanceof Max) {
        bucket.setMax(max(a));
    } else if (a instanceof Sum) {
        bucket.setSum(sum(a));
    } else if (a instanceof Stats) {
        stats(bucket, a);
    }  else if (a instanceof ValueCount) {
        bucket.setValueCount(count(a));
    } else {
        throw new UnsupportedOperationException("不支持的聚合类型");
    }
}
 

@Override
TotalSalesQuery.Result extractResult(SearchResponse searchResponse) {
    TotalSalesQuery.Result.ResultBuilder result = TotalSalesQuery.Result.builder();
    long count = 0;
    BigDecimal amount = BigDecimal.ZERO;
    Filter filterAgg = searchResponse.getAggregations().get("agg_filter");
    Terms products = filterAgg.getAggregations().get("count_by_product");
    for (Terms.Bucket b : products.getBuckets()){
        count += b.getDocCount();
        Sum sum = b.getAggregations().get("total_premium");
        amount = amount.add(BigDecimal.valueOf(sum.getValue()).setScale(2,BigDecimal.ROUND_HALF_UP));
        result.productTotal(b.getKeyAsString(), SalesResult.of(b.getDocCount(),BigDecimal.valueOf(sum.getValue())));
    }
    result.total(SalesResult.of(count,amount));

    return result.build();
}
 

@Override
SalesTrendsQuery.Result extractResult(SearchResponse searchResponse) {
    SalesTrendsQuery.Result.ResultBuilder result = SalesTrendsQuery.Result.builder();

    Filter filterAgg = searchResponse.getAggregations().get("agg_filter");
    Histogram agg = filterAgg.getAggregations().get("sales");
    for (Histogram.Bucket b : agg.getBuckets()){
        DateTime key = (DateTime)b.getKey();
        Sum sum = b.getAggregations().get("total_premium");
        result.periodSale(
                new SalesTrendsQuery.PeriodSales(
                        LocalDate.of(key.getYear(),key.getMonthOfYear(),key.getDayOfMonth()),
                        b.getKeyAsString(),
                        SalesResult.of(b.getDocCount(), BigDecimal.valueOf(sum.getValue()).setScale(2, BigDecimal.ROUND_HALF_UP))
                )
        );
    }

    return result.build();
}
 

@Override
AgentSalesQuery.Result extractResult(SearchResponse searchResponse) {
    AgentSalesQuery.Result.ResultBuilder result = AgentSalesQuery.Result.builder();
    Filter filterAgg = searchResponse.getAggregations().get("agg_filter");
    Terms agents = filterAgg.getAggregations().get("count_by_agent");

    for (Terms.Bucket b : agents.getBuckets()) {
        Sum sum = b.getAggregations().get("total_premium");
        result.agentTotal(
                b.getKeyAsString(),
                SalesResult.of(b.getDocCount(), BigDecimal.valueOf(sum.getValue()))
        );
    }

    return result.build();
}
 

private List<GroupResult> getGroupResults(GroupRequest groupRequest, int index, Aggregations aggregations, Map<String, FieldType> commonColumnMetadata) {
  List<Group> groups = groupRequest.getGroups();
  String field = groups.get(index).getField();
  List<GroupResult> searchResultGroups = new ArrayList<>();
  if(aggregations != null) {
    Terms terms = aggregations.get(getGroupByAggregationName(field));
    for (Bucket bucket : terms.getBuckets()) {
      GroupResult groupResult = new GroupResult();
      groupResult.setKey(formatKey(bucket.getKey(), commonColumnMetadata.get(field)));
      groupResult.setTotal(bucket.getDocCount());
      Optional<String> scoreField = groupRequest.getScoreField();
      if (scoreField.isPresent()) {
        Sum score = bucket.getAggregations().get(getSumAggregationName(scoreField.get()));
        groupResult.setScore(score.getValue());
      }
      if (index < groups.size() - 1) {
        groupResult.setGroupedBy(groups.get(index + 1).getField());
        groupResult.setGroupResults(getGroupResults(groupRequest, index + 1, bucket.getAggregations(), commonColumnMetadata));
      }
      searchResultGroups.add(groupResult);
    }
  }
  return searchResultGroups;
}
 

private long getEntitySizeAggregation( final SearchRequestBuilder builder ) {
    final String key = "entitySize";
    SumBuilder sumBuilder = new SumBuilder(key);
    sumBuilder.field("entitySize");
    builder.addAggregation(sumBuilder);

    Observable<Number> o = Observable.from(builder.execute())
        .map(response -> {
            Sum aggregation = (Sum) response.getAggregations().get(key);
            if(aggregation == null){
                return -1;
            }else{
                return aggregation.getValue();
            }
        });
    Number val =   ObservableTimer.time(o,aggregationTimer).toBlocking().lastOrDefault(-1);
    return val.longValue();
}
 

@Override
public int readMetricsValue(final MetricsCondition condition,
                            final String valueColumnName,
                            final Duration duration) throws IOException {
    SearchSourceBuilder sourceBuilder = SearchSourceBuilder.searchSource();
    buildQuery(sourceBuilder, condition, duration);

    TermsAggregationBuilder entityIdAggregation = AggregationBuilders.terms(Metrics.ENTITY_ID)
                                                                     .field(Metrics.ENTITY_ID)
                                                                     .size(1);
    final Function function = ValueColumnMetadata.INSTANCE.getValueFunction(condition.getName());
    functionAggregation(function, entityIdAggregation, valueColumnName);

    sourceBuilder.aggregation(entityIdAggregation);

    SearchResponse response = getClient().search(condition.getName(), sourceBuilder);

    Terms idTerms = response.getAggregations().get(Metrics.ENTITY_ID);
    for (Terms.Bucket idBucket : idTerms.getBuckets()) {
        switch (function) {
            case Sum:
                Sum sum = idBucket.getAggregations().get(valueColumnName);
                return (int) sum.getValue();
            case Avg:
                Avg avg = idBucket.getAggregations().get(valueColumnName);
                return (int) avg.getValue();
            default:
                avg = idBucket.getAggregations().get(valueColumnName);
                return (int) avg.getValue();
        }
    }
    return ValueColumnMetadata.INSTANCE.getDefaultValue(condition.getName());
}
 

protected void functionAggregation(Function function, TermsAggregationBuilder parentAggBuilder, String valueCName) {
    switch (function) {
        case Avg:
            parentAggBuilder.subAggregation(AggregationBuilders.avg(valueCName).field(valueCName));
            break;
        case Sum:
            parentAggBuilder.subAggregation(AggregationBuilders.sum(valueCName).field(valueCName));
            break;
        default:
            parentAggBuilder.subAggregation(AggregationBuilders.avg(valueCName).field(valueCName));
            break;
    }
}
 

public static String getValueFromAggregation(Aggregation a, Function f){
	
	String value = null;
	switch(f.type){
	case Function.SUM :
		value = String.valueOf(((Sum) a).getValue());
		break;
	case Function.COUNT :
		value = String.valueOf(((ValueCount) a).getValue());
		break;
	case Function.DC :
		value = String.valueOf(((Cardinality) a).getValue());
		break;
	case Function.AVG :
		value = String.valueOf(((Avg) a).getValue());
		break;
	case Function.MAX :
		value = String.valueOf(((Max) a).getValue());
		break;
	case Function.MIN :
		value = String.valueOf(((Min) a).getValue());
		break;
	}
	
	return value;
	
	
}