ITQC/gls_docs
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ATEST-315

Browse Source
This commit is contained in:
Fabian Märki
2016-04-14 14:54:22 +02:00
parent 659c4d5f3a
commit 095348d405
10 changed files with 816 additions and 621 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java
+9
View File
@@ -2,6 +2,7 @@ package ch.psi.daq.queryrest.controller;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
@@ -17,6 +18,7 @@ import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import javax.validation.Valid;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.slf4j.Logger;
@@ -39,6 +41,7 @@ import ch.psi.daq.cassandra.request.validate.RequestProviderValidator;
import ch.psi.daq.common.json.deserialize.AttributeBasedDeserializer;
import ch.psi.daq.common.ordering.Ordering;
import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.cassandra.FieldNames;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.reader.Backend;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
@@ -61,6 +64,7 @@ import ch.psi.daq.queryrest.response.json.JSONResponseStreamWriter;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import com.hazelcast.nio.serialization.FieldType;
@RestController
public class QueryRestController {
@@ -281,6 +285,11 @@ public class QueryRestController {
LOGGER.warn(message);
throw new IllegalArgumentException(message);
}
if(!ArrayUtils.contains(query.getColumns(), FieldNames.FIELD_GLOBAL_TIME)){
query.addField(QueryField.globalMillis);
}
}
try {
src/main/java/ch/psi/daq/queryrest/response/csv/AggregationStringifyer.java
+25
View File
@@ -0,0 +1,25 @@
package ch.psi.daq.queryrest.response.csv;
import java.util.function.Function;
import java.util.function.ToDoubleFunction;
import ch.psi.daq.domain.DataEvent;
public class AggregationStringifyer implements Function<DataEvent, String> {
private ToDoubleFunction<DataEvent> accessor;
private String nonValue;
public AggregationStringifyer(ToDoubleFunction<DataEvent> accessor, String nonValue) {
this.accessor = accessor;
this.nonValue = nonValue;
}
@Override
public String apply(DataEvent event) {
if (event == null) {
return nonValue;
}else{
return Double.toString(accessor.applyAsDouble(event));
}
}
}
src/main/java/ch/psi/daq/queryrest/response/csv/ArrayProcessor.java
-57
View File
@@ -1,57 +0,0 @@
package ch.psi.daq.queryrest.response.csv;
import java.lang.reflect.Array;
import org.supercsv.cellprocessor.CellProcessorAdaptor;
import org.supercsv.cellprocessor.ift.CellProcessor;
import org.supercsv.util.CsvContext;
public class ArrayProcessor extends CellProcessorAdaptor {
public static final char DEFAULT_SEPARATOR = ',';
public static final String OPEN_BRACKET = "[";
public static final String CLOSE_BRACKET = "]";
private char separator = DEFAULT_SEPARATOR;
public ArrayProcessor() {
super();
}
public ArrayProcessor(char separator) {
super();
this.separator = separator;
}
public ArrayProcessor(CellProcessor next) {
super(next);
}
public ArrayProcessor(CellProcessor next, char separator) {
super(next);
this.separator = separator;
}
@Override
public Object execute(Object value, CsvContext context) {
if (value.getClass().isArray()) {
StringBuilder buf = new StringBuilder();
int length = Array.getLength(value);
buf.append(OPEN_BRACKET);
for (int i = 0; i < length;) {
Object val = next.execute(Array.get(value, i), context);
buf.append(val);
++i;
if (i < length) {
buf.append(separator);
}
}
buf.append(CLOSE_BRACKET);
return buf.toString();
} else {
return next.execute(value, context);
}
}
}
src/main/java/ch/psi/daq/queryrest/response/csv/CSVResponseStreamWriter.java
+94 -124
View File
@@ -1,32 +1,39 @@
package ch.psi.daq.queryrest.response.csv;
import java.io.BufferedWriter;
import java.io.IOException;
import java.io.OutputStream;
import java.io.OutputStreamWriter;
import java.io.Writer;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.LinkedHashSet;
import java.util.Collection;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import java.util.function.ToLongFunction;
import java.util.stream.Stream;
import javax.annotation.Resource;
import javax.servlet.ServletResponse;
import javax.servlet.http.HttpServletResponse;
import org.apache.commons.csv.CSVFormat;
import org.apache.commons.csv.CSVPrinter;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.supercsv.cellprocessor.constraint.NotNull;
import org.supercsv.cellprocessor.ift.CellProcessor;
import org.supercsv.io.dozer.CsvDozerBeanWriter;
import org.supercsv.io.dozer.ICsvDozerBeanWriter;
import org.supercsv.prefs.CsvPreference;
import com.fasterxml.jackson.core.JsonEncoding;
import ch.psi.daq.common.stream.StreamIterable;
import ch.psi.daq.common.stream.StreamMatcher;
import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
@@ -44,8 +51,15 @@ import ch.psi.daq.queryrest.response.AbstractResponseStreamWriter;
*/
public class CSVResponseStreamWriter extends AbstractResponseStreamWriter {
private static final char DELIMITER_CVS = ';';
private static final char DELIMITER_ARRAY = ',';
public static final char DELIMITER_CVS = ';';
public static final String DELIMITER_ARRAY = ",";
public static final char DELIMITER_CHANNELNAME_FIELDNAME = '.';
public static final String EMPTY_VALUE = "";
private static final Function<Pair<ChannelName, DataEvent>, ChannelName> KEY_PROVIDER = (pair) -> pair.getKey();
// try to match sync data (bsread) with non sync data (epics) based on the time usin 10 millis
// buckets.
private static final ToLongFunction<Pair<ChannelName, DataEvent>> MATCHER_PROVIDER = (pair) -> pair.getValue()
.getGlobalMillis() / 10L;
private static final Logger LOGGER = LoggerFactory.getLogger(CSVResponseStreamWriter.class);
@@ -65,60 +79,82 @@ public class CSVResponseStreamWriter extends AbstractResponseStreamWriter {
private void respondInternal(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
ResponseOptions options, HttpServletResponse response) throws Exception {
AtomicReference<Exception> exception = new AtomicReference<>();
OutputStream out = handleCompressionAndResponseHeaders(options, response, CONTENT_TYPE_CSV);
List<ICsvDozerBeanWriter> beanWriters = new ArrayList<>();
final OutputStream out = handleCompressionAndResponseHeaders(options, response, CONTENT_TYPE_CSV);
final Map<ChannelName, Stream<Pair<ChannelName, DataEvent>>> streams = new LinkedHashMap<>(results.size());
final List<String> header = new ArrayList<>();
final Collection<Pair<ChannelName, Function<DataEvent, String>>> accessors = new ArrayList<>();
// prepare the streams
results.forEach(entry -> {
DAQQueryElement query = entry.getKey();
Set<QueryField> queryFields = query.getFields();
List<Aggregation> aggregations = query.getAggregations();
int aggregationsSize = (aggregations != null ? aggregations.size() : 0);
Set<String> fieldMapping = new LinkedHashSet<>(queryFields.size() + aggregationsSize);
List<String> header = new ArrayList<>(queryFields.size() + aggregationsSize);
AtomicReference<CellProcessor[]> processorsRef = new AtomicReference<>();
final DAQQueryElement query = entry.getKey();
entry.getValue()
.sequential()
.forEach(triple -> {
try {
CellProcessor[] processors = processorsRef.get();
ICsvDozerBeanWriter beanWriter;
if (triple.getRight() instanceof Stream) {
setupChannelColumns(query, triple.getLeft(), triple.getMiddle(), header, accessors);
if (processors == null) {
processors = setupCellProcessors(query, triple.getLeft(), fieldMapping, header);
processorsRef.set(processors);
beanWriter = setupBeanWriter(fieldMapping, out);
beanWriters.add(beanWriter);
} else {
beanWriter = beanWriters.get(beanWriters.size() - 1);
}
writeToOutput(triple, processors, beanWriter, header);
} catch (Exception e) {
LOGGER.warn("Could not write CSV of '{}'", triple.getMiddle(), e);
exception.compareAndSet(null, e);
final Stream<Pair<ChannelName, DataEvent>> eventStream = ((Stream<DataEvent>) triple.getRight())
.map(dataEvent -> Pair.of(triple.getMiddle(), dataEvent));
streams.put(triple.getMiddle(), eventStream);
} else {
final String message = String.format("Expect a DataEvent Stream for '%s'.", triple.getMiddle());
LOGGER.warn(message);
}
});
if (!beanWriters.isEmpty()) {
try {
beanWriters.get(beanWriters.size() - 1).flush();
} catch (Exception e) {
LOGGER.error("Could not flush ICsvDozerBeanWriter.", e);
exception.compareAndSet(null, e);
}
}
});
for (ICsvDozerBeanWriter beanWriter : beanWriters) {
// online matching of the stream's content
StreamMatcher<ChannelName, Pair<ChannelName, DataEvent>> streamMatcher =
new StreamMatcher<>(KEY_PROVIDER, MATCHER_PROVIDER, streams.values());
Iterator<Map<ChannelName, Pair<ChannelName, DataEvent>>> streamsMatchIter = streamMatcher.iterator();
// prepare csv output
CSVFormat csvFormat = CSVFormat.EXCEL.withDelimiter(DELIMITER_CVS);
Writer writer = null;
CSVPrinter csvFilePrinter = null;
try {
writer = new BufferedWriter(new OutputStreamWriter(out, StandardCharsets.UTF_8));
// writer = new BufferedWriter(new OutputStreamWriter(out,
// response.getCharacterEncoding()));
csvFilePrinter = new CSVPrinter(writer, csvFormat);
csvFilePrinter.printRecord(header);
while (streamsMatchIter.hasNext()) {
final Map<ChannelName, Pair<ChannelName, DataEvent>> match = streamsMatchIter.next();
// ensure correct order
Stream<String> rowStream = accessors.stream().sequential()
.map(accessorPair -> {
Pair<ChannelName, DataEvent> eventPair = match.get(accessorPair.getKey());
if (eventPair != null) {
return accessorPair.getValue().apply(eventPair.getValue());
} else {
return EMPTY_VALUE;
}
});
csvFilePrinter.printRecord(new StreamIterable<String>(rowStream));
}
} catch (IOException e) {
LOGGER.error("Could not write CSV.", e);
exception.compareAndSet(null, e);
} finally {
try {
beanWriter.close();
} catch (Exception e) {
LOGGER.error("Could not close ICsvDozerBeanWriter.", e);
if (writer != null) {
writer.flush();
writer.close();
}
if (csvFilePrinter != null) {
csvFilePrinter.close();
}
} catch (IOException e) {
LOGGER.error("Could not close CSV writer.", e);
exception.compareAndSet(null, e);
}
}
@@ -128,95 +164,29 @@ public class CSVResponseStreamWriter extends AbstractResponseStreamWriter {
}
}
/**
* Sets up the bean writer instance.
*
* @throws Exception
*
*/
private ICsvDozerBeanWriter setupBeanWriter(Set<String> fieldMapping, OutputStream out) throws Exception {
CsvPreference preference = new CsvPreference.Builder(
(char) CsvPreference.EXCEL_NORTH_EUROPE_PREFERENCE.getQuoteChar(),
DELIMITER_CVS,
CsvPreference.EXCEL_NORTH_EUROPE_PREFERENCE.getEndOfLineSymbols()).build();
ICsvDozerBeanWriter beanWriter = new CsvDozerBeanWriter(new OutputStreamWriter(out), preference);
// configure the mapping from the fields to the CSV columns
beanWriter.configureBeanMapping(DataEvent.class, fieldMapping.toArray(new String[fieldMapping.size()]));
return beanWriter;
}
/**
* Sets up the array of {@link CellProcessor}s needed for later configuration of the bean writer.
*
* Cell processors are an integral part of reading and writing with Super CSV - they automate the
* data type conversions, and enforce constraints. They implement the chain of responsibility
* design pattern - each processor has a single, well-defined purpose and can be chained together
* with other processors to fully automate all of the required conversions and constraint
* validation for a single CSV column.
*
* @param daqQuery The current {@link DAQQueryElement}
* @param backendQuery One BackendQuery of the current {@link DAQQueryElement}
* @return Array of {@link CellProcessor} entries
*/
private CellProcessor[] setupCellProcessors(DAQQueryElement daqQuery, BackendQuery backendQuery,
Set<String> fieldMapping, List<String> header) {
private void setupChannelColumns(DAQQueryElement daqQuery, BackendQuery backendQuery, ChannelName channelName,
Collection<String> header, Collection<Pair<ChannelName, Function<DataEvent, String>>> accessors) {
Set<QueryField> queryFields = daqQuery.getFields();
List<Aggregation> aggregations = daqQuery.getAggregations();
List<CellProcessor> processorSet =
new ArrayList<>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
boolean isNewField;
QueryAnalyzer queryAnalyzer = queryAnalizerFactory.apply(backendQuery);
for (QueryField field : queryFields) {
if (!(QueryField.value.equals(field) && queryAnalyzer.isAggregationEnabled())) {
isNewField = fieldMapping.add(field.name());
header.add(channelName.getName() + DELIMITER_CHANNELNAME_FIELDNAME + field.name());
if (isNewField) {
header.add(field.name());
processorSet.add(new ArrayProcessor(new NotNull(), DELIMITER_ARRAY));
}
accessors.add(Pair.of(channelName, new QueryFieldStringifyer(field.getAccessor(), EMPTY_VALUE,
DELIMITER_ARRAY)));
}
}
if (aggregations != null && queryAnalyzer.isAggregationEnabled()) {
for (Aggregation aggregation : daqQuery.getAggregations()) {
isNewField = fieldMapping.add("value." + aggregation.name());
if (isNewField) {
header.add(aggregation.name());
processorSet.add(new ArrayProcessor(new NotNull(), DELIMITER_ARRAY));
}
header.add(channelName.getName() + DELIMITER_CHANNELNAME_FIELDNAME + QueryField.value.name()
+ DELIMITER_CHANNELNAME_FIELDNAME + aggregation.name());
accessors.add(Pair.of(channelName, new AggregationStringifyer(aggregation.getAccessor(), EMPTY_VALUE)));
}
}
return processorSet.toArray(new CellProcessor[processorSet.size()]);
}
@SuppressWarnings("unchecked")
private void writeToOutput(Triple<BackendQuery, ChannelName, ?> triple, CellProcessor[] processors,
ICsvDozerBeanWriter beanWriter, List<String> header) throws IOException {
if (triple.getRight() instanceof Stream) {
beanWriter.writeComment("");
beanWriter.writeComment("Start of " + triple.getMiddle());
beanWriter.writeHeader(header.toArray(new String[header.size()]));
Stream<DataEvent> eventStream = (Stream<DataEvent>) triple.getRight();
eventStream
.forEach(
event -> {
try {
beanWriter.write(event, processors);
} catch (Exception e) {
LOGGER.error("Could not write elements for '{}-{}'", event.getChannel(),
event.getPulseId(), e);
}
});
} else {
String message = "Type '" + triple.getRight().getClass() + "' not supported.";
LOGGER.error(message);
}
}
}
src/main/java/ch/psi/daq/queryrest/response/csv/QueryFieldStringifyer.java
+51
View File
@@ -0,0 +1,51 @@
package ch.psi.daq.queryrest.response.csv;
import java.lang.reflect.Array;
import java.util.function.Function;
import ch.psi.daq.domain.DataEvent;
public class QueryFieldStringifyer implements Function<DataEvent, String> {
public static final String OPEN_BRACKET = "[";
public static final String CLOSE_BRACKET = "]";
private Function<DataEvent, Object> accessor;
private String nonValue;
private String arraySeparator;
public QueryFieldStringifyer(Function<DataEvent, Object> accessor, String nonValue, String arraySeparator) {
this.accessor = accessor;
this.nonValue = nonValue;
this.arraySeparator = arraySeparator;
}
@Override
public String apply(DataEvent event) {
if (event == null) {
return nonValue;
}
Object value = accessor.apply(event);
if (value == null) {
return nonValue;
} else if (value.getClass().isArray()) {
StringBuilder buf = new StringBuilder();
int length = Array.getLength(value);
buf.append(OPEN_BRACKET);
for (int i = 0; i < length;) {
buf.append(Array.get(value, i));
++i;
if (i < length) {
buf.append(arraySeparator);
}
}
buf.append(CLOSE_BRACKET);
return buf.toString();
} else {
return value.toString();
}
}
}