src/main/java/org/influxdb/impl/InfluxDBResultMapper.java
/*
* The MIT License (MIT)
*
* Copyright (c) 2017 azeti Networks AG (<info@azeti.net>)
*
* Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
* associated documentation files (the "Software"), to deal in the Software without restriction,
* including without limitation the rights to use, copy, modify, merge, publish, distribute,
* sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is
* furnished to do so, subject to the following conditions:
*
* The above copyright notice and this permission notice shall be included in all copies or
* substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT
* NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM,
* DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
package org.influxdb.impl;
import org.influxdb.InfluxDBMapperException;
import org.influxdb.annotation.Column;
import org.influxdb.annotation.Exclude;
import org.influxdb.annotation.Measurement;
import org.influxdb.dto.QueryResult;
import java.lang.reflect.Field;
import java.lang.reflect.Modifier;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.time.Instant;
import java.time.format.DateTimeFormatter;
import java.time.format.DateTimeFormatterBuilder;
import java.time.temporal.ChronoField;
import java.util.LinkedList;
import java.util.List;
import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.TimeUnit;
/**
* Main class responsible for mapping a QueryResult to a POJO.
*
* @author fmachado
*/
public class InfluxDBResultMapper {
/**
* Data structure used to cache classes used as measurements.
*/
private static class ClassInfo {
ConcurrentMap<String, Field> fieldMap;
ConcurrentMap<Field, TypeMapper> typeMappers;
}
private static final
ConcurrentMap<String, ClassInfo> CLASS_INFO_CACHE = new ConcurrentHashMap<>();
private static final int FRACTION_MIN_WIDTH = 0;
private static final int FRACTION_MAX_WIDTH = 9;
private static final boolean ADD_DECIMAL_POINT = true;
/**
* When a query is executed without {@link TimeUnit}, InfluxDB returns the <code>time</code>
* column as a RFC3339 date.
*/
private static final DateTimeFormatter RFC3339_FORMATTER = new DateTimeFormatterBuilder()
.appendPattern("yyyy-MM-dd'T'HH:mm:ss")
.appendFraction(ChronoField.NANO_OF_SECOND, FRACTION_MIN_WIDTH, FRACTION_MAX_WIDTH, ADD_DECIMAL_POINT)
.appendZoneOrOffsetId()
.toFormatter();
/**
* <p>
* Process a {@link QueryResult} object returned by the InfluxDB client inspecting the internal
* data structure and creating the respective object instances based on the Class passed as
* parameter.
* </p>
*
* @param queryResult the InfluxDB result object
* @param clazz the Class that will be used to hold your measurement data
* @param <T> the target type
*
* @return a {@link List} of objects from the same Class passed as parameter and sorted on the
* same order as received from InfluxDB.
*
* @throws InfluxDBMapperException If {@link QueryResult} parameter contain errors,
* <code>clazz</code> parameter is not annotated with @Measurement or it was not
* possible to define the values of your POJO (e.g. due to an unsupported field type).
*/
public <T> List<T> toPOJO(final QueryResult queryResult, final Class<T> clazz) throws InfluxDBMapperException {
return toPOJO(queryResult, clazz, TimeUnit.MILLISECONDS);
}
/**
* <p>
* Process a {@link QueryResult} object returned by the InfluxDB client inspecting the internal
* data structure and creating the respective object instances based on the Class passed as
* parameter.
* </p>
*
* @param queryResult the InfluxDB result object
* @param clazz the Class that will be used to hold your measurement data
* @param precision the time precision of results
* @param <T> the target type
*
* @return a {@link List} of objects from the same Class passed as parameter and sorted on the
* same order as received from InfluxDB.
*
* @throws InfluxDBMapperException If {@link QueryResult} parameter contain errors,
* <code>clazz</code> parameter is not annotated with @Measurement or it was not
* possible to define the values of your POJO (e.g. due to an unsupported field type).
*/
public <T> List<T> toPOJO(final QueryResult queryResult, final Class<T> clazz,
final TimeUnit precision) throws InfluxDBMapperException {
throwExceptionIfMissingAnnotation(clazz);
String measurementName = getMeasurementName(clazz);
return this.toPOJO(queryResult, clazz, measurementName, precision);
}
/**
* <p>
* Process a {@link QueryResult} object returned by the InfluxDB client inspecting the internal
* data structure and creating the respective object instances based on the Class passed as
* parameter.
* </p>
*
* @param queryResult the InfluxDB result object
* @param clazz the Class that will be used to hold your measurement data
* @param <T> the target type
* @param measurementName name of the Measurement
*
* @return a {@link List} of objects from the same Class passed as parameter and sorted on the
* same order as received from InfluxDB.
*
* @throws InfluxDBMapperException If {@link QueryResult} parameter contain errors,
* <code>clazz</code> parameter is not annotated with @Measurement or it was not
* possible to define the values of your POJO (e.g. due to an unsupported field type).
*/
public <T> List<T> toPOJO(final QueryResult queryResult, final Class<T> clazz, final String measurementName)
throws InfluxDBMapperException {
return toPOJO(queryResult, clazz, measurementName, TimeUnit.MILLISECONDS);
}
/**
* <p>
* Process a {@link QueryResult} object returned by the InfluxDB client inspecting the internal
* data structure and creating the respective object instances based on the Class passed as
* parameter.
* </p>
*
* @param queryResult the InfluxDB result object
* @param clazz the Class that will be used to hold your measurement data
* @param <T> the target type
* @param measurementName name of the Measurement
* @param precision the time precision of results
*
* @return a {@link List} of objects from the same Class passed as parameter and sorted on the
* same order as received from InfluxDB.
*
* @throws InfluxDBMapperException If {@link QueryResult} parameter contain errors,
* <code>clazz</code> parameter is not annotated with @Measurement or it was not
* possible to define the values of your POJO (e.g. due to an unsupported field type).
*/
public <T> List<T> toPOJO(final QueryResult queryResult, final Class<T> clazz, final String measurementName,
final TimeUnit precision)
throws InfluxDBMapperException {
Objects.requireNonNull(measurementName, "measurementName");
Objects.requireNonNull(queryResult, "queryResult");
Objects.requireNonNull(clazz, "clazz");
throwExceptionIfResultWithError(queryResult);
cacheMeasurementClass(clazz);
List<T> result = new LinkedList<T>();
queryResult.getResults().stream()
.filter(internalResult -> Objects.nonNull(internalResult) && Objects.nonNull(internalResult.getSeries()))
.forEach(internalResult -> {
internalResult.getSeries().stream()
.filter(series -> series.getName().equals(measurementName))
.forEachOrdered(series -> {
parseSeriesAs(series, clazz, result, precision);
});
});
return result;
}
void throwExceptionIfMissingAnnotation(final Class<?> clazz) {
if (!clazz.isAnnotationPresent(Measurement.class)) {
throw new IllegalArgumentException(
"Class " + clazz.getName() + " is not annotated with @" + Measurement.class.getSimpleName());
}
}
void throwExceptionIfResultWithError(final QueryResult queryResult) {
if (queryResult.getError() != null) {
throw new InfluxDBMapperException("InfluxDB returned an error: " + queryResult.getError());
}
queryResult.getResults().forEach(seriesResult -> {
if (seriesResult.getError() != null) {
throw new InfluxDBMapperException("InfluxDB returned an error with Series: " + seriesResult.getError());
}
});
}
void cacheMeasurementClass(final Class<?>... classVarAgrs) {
for (Class<?> clazz : classVarAgrs) {
if (CLASS_INFO_CACHE.containsKey(clazz.getName())) {
continue;
}
ConcurrentMap<String, Field> fieldMap = new ConcurrentHashMap<>();
ConcurrentMap<Field, TypeMapper> typeMappers = new ConcurrentHashMap<>();
Measurement measurement = clazz.getAnnotation(Measurement.class);
boolean allFields = measurement != null && measurement.allFields();
Class<?> c = clazz;
TypeMapper typeMapper = TypeMapper.empty();
while (c != null) {
for (Field field : c.getDeclaredFields()) {
Column colAnnotation = field.getAnnotation(Column.class);
if (colAnnotation == null && !(allFields
&& !field.isAnnotationPresent(Exclude.class) && !Modifier.isStatic(field.getModifiers()))) {
continue;
}
fieldMap.put(getFieldName(field, colAnnotation), field);
typeMappers.put(field, typeMapper);
}
Class<?> superclass = c.getSuperclass();
Type genericSuperclass = c.getGenericSuperclass();
if (genericSuperclass instanceof ParameterizedType) {
typeMapper = TypeMapper.of((ParameterizedType) genericSuperclass, superclass);
} else {
typeMapper = TypeMapper.empty();
}
c = superclass;
}
ClassInfo classInfo = new ClassInfo();
classInfo.fieldMap = fieldMap;
classInfo.typeMappers = typeMappers;
CLASS_INFO_CACHE.putIfAbsent(clazz.getName(), classInfo);
}
}
private static String getFieldName(final Field field, final Column colAnnotation) {
if (colAnnotation != null && !colAnnotation.name().isEmpty()) {
return colAnnotation.name();
}
return field.getName();
}
String getMeasurementName(final Class<?> clazz) {
return ((Measurement) clazz.getAnnotation(Measurement.class)).name();
}
String getDatabaseName(final Class<?> clazz) {
return ((Measurement) clazz.getAnnotation(Measurement.class)).database();
}
String getRetentionPolicy(final Class<?> clazz) {
return ((Measurement) clazz.getAnnotation(Measurement.class)).retentionPolicy();
}
<T> List<T> parseSeriesAs(final QueryResult.Series series, final Class<T> clazz, final List<T> result) {
return parseSeriesAs(series, clazz, result, TimeUnit.MILLISECONDS);
}
<T> List<T> parseSeriesAs(final QueryResult.Series series, final Class<T> clazz, final List<T> result,
final TimeUnit precision) {
int columnSize = series.getColumns().size();
ClassInfo classInfo = CLASS_INFO_CACHE.get(clazz.getName());
try {
T object = null;
for (List<Object> row : series.getValues()) {
for (int i = 0; i < columnSize; i++) {
Field correspondingField = classInfo.fieldMap.get(series.getColumns().get(i)/*InfluxDB columnName*/);
if (correspondingField != null) {
if (object == null) {
object = clazz.newInstance();
}
setFieldValue(object, correspondingField, row.get(i), precision,
classInfo.typeMappers.get(correspondingField));
}
}
// When the "GROUP BY" clause is used, "tags" are returned as Map<String,String> and
// accordingly with InfluxDB documentation
// https://docs.influxdata.com/influxdb/v1.2/concepts/glossary/#tag-value
// "tag" values are always String.
if (series.getTags() != null && !series.getTags().isEmpty()) {
for (Entry<String, String> entry : series.getTags().entrySet()) {
Field correspondingField = classInfo.fieldMap.get(entry.getKey()/*InfluxDB columnName*/);
if (correspondingField != null) {
// I don't think it is possible to reach here without a valid "object"
setFieldValue(object, correspondingField, entry.getValue(), precision,
classInfo.typeMappers.get(correspondingField));
}
}
}
if (object != null) {
result.add(object);
object = null;
}
}
} catch (InstantiationException | IllegalAccessException e) {
throw new InfluxDBMapperException(e);
}
return result;
}
/**
* InfluxDB client returns any number as Double.
* See <a href="https://github.com/influxdata/influxdb-java/issues/153#issuecomment-259681987">...</a>
* for more information.
*
*/
private static <T> void setFieldValue(final T object, final Field field, final Object value, final TimeUnit precision,
final TypeMapper typeMapper)
throws IllegalArgumentException, IllegalAccessException {
if (value == null) {
return;
}
Type fieldType = typeMapper.resolve(field.getGenericType());
if (!field.isAccessible()) {
field.setAccessible(true);
}
field.set(object, adaptValue((Class<?>) fieldType, value, precision, field.getName(), object.getClass().getName()));
}
private static Object adaptValue(final Class<?> fieldType, final Object value, final TimeUnit precision,
final String fieldName, final String className) {
try {
if (String.class.isAssignableFrom(fieldType)) {
return String.valueOf(value);
}
if (Instant.class.isAssignableFrom(fieldType)) {
if (value instanceof String) {
return Instant.from(RFC3339_FORMATTER.parse(String.valueOf(value)));
}
if (value instanceof Long) {
return Instant.ofEpochMilli(toMillis((long) value, precision));
}
if (value instanceof Double) {
return Instant.ofEpochMilli(toMillis(((Double) value).longValue(), precision));
}
if (value instanceof Integer) {
return Instant.ofEpochMilli(toMillis(((Integer) value).longValue(), precision));
}
throw new InfluxDBMapperException("Unsupported type " + fieldType + " for field " + fieldName);
}
if (Double.class.isAssignableFrom(fieldType) || double.class.isAssignableFrom(fieldType)) {
return value;
}
if (Long.class.isAssignableFrom(fieldType) || long.class.isAssignableFrom(fieldType)) {
return ((Double) value).longValue();
}
if (Integer.class.isAssignableFrom(fieldType) || int.class.isAssignableFrom(fieldType)) {
return ((Double) value).intValue();
}
if (Boolean.class.isAssignableFrom(fieldType) || boolean.class.isAssignableFrom(fieldType)) {
return Boolean.valueOf(String.valueOf(value));
}
if (Enum.class.isAssignableFrom(fieldType)) {
//noinspection unchecked
return Enum.valueOf((Class<Enum>) fieldType, String.valueOf(value));
}
} catch (ClassCastException e) {
String msg = "Class '%s' field '%s' was defined with a different field type and caused a ClassCastException. "
+ "The correct type is '%s' (current field value: '%s').";
throw new InfluxDBMapperException(
String.format(msg, className, fieldName, value.getClass().getName(), value));
}
throw new InfluxDBMapperException(
String.format("Class '%s' field '%s' is from an unsupported type '%s'.", className, fieldName, fieldType));
}
private static long toMillis(final long value, final TimeUnit precision) {
return TimeUnit.MILLISECONDS.convert(value, precision);
}
}