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

ATEST-594

Browse Source
This commit is contained in:
Fabian Märki
2016-11-04 17:17:52 +01:00
parent 3d920ef048
commit 6dbd79bbd0
17 changed files with 2580 additions and 166 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java
+7 -1
View File
@@ -33,8 +33,8 @@ import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.backend.analyzer.BackendQueryAnalyzer;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.operation.Response;
import ch.psi.daq.domain.query.operation.aggregation.extrema.AbstractExtremaMeta;
import ch.psi.daq.domain.query.response.Response;
import ch.psi.daq.query.analyzer.BackendQueryAnalyzerImpl;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.queryrest.controller.validator.QueryValidator;
@@ -44,6 +44,7 @@ import ch.psi.daq.queryrest.query.QueryManagerImpl;
import ch.psi.daq.queryrest.response.PolymorphicResponseMixIn;
import ch.psi.daq.queryrest.response.csv.CSVResponseStreamWriter;
import ch.psi.daq.queryrest.response.json.JSONResponseStreamWriter;
import ch.psi.daq.queryrest.response.json.JSONTableResponseStreamWriter;
@Configuration
@Import(value = QueryRestConfigCORS.class)
@@ -128,6 +129,11 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
public JSONResponseStreamWriter jsonResponseStreamWriter() {
return new JSONResponseStreamWriter();
}
@Bean
public JSONTableResponseStreamWriter jsonTableResponseStreamWriter() {
return new JSONTableResponseStreamWriter();
}
@Bean
public CSVResponseStreamWriter csvResponseStreamWriter() {
src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java
+2 -2
View File
@@ -39,8 +39,8 @@ import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.AggregationType;
import ch.psi.daq.domain.query.operation.Compression;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.operation.Response;
import ch.psi.daq.domain.query.operation.ResponseFormat;
import ch.psi.daq.domain.query.response.Response;
import ch.psi.daq.domain.query.response.ResponseFormat;
import ch.psi.daq.domain.request.validate.RequestProviderValidator;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
src/main/java/ch/psi/daq/queryrest/response/AbstractHTTPResponse.java
+2 -2
View File
@@ -10,8 +10,8 @@ import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.core.JsonEncoding;
import ch.psi.daq.domain.query.DAQQueries;
import ch.psi.daq.domain.query.operation.ResponseFormat;
import ch.psi.daq.domain.query.operation.ResponseImpl;
import ch.psi.daq.domain.query.response.ResponseFormat;
import ch.psi.daq.domain.query.response.ResponseImpl;
public abstract class AbstractHTTPResponse extends ResponseImpl {
src/main/java/ch/psi/daq/queryrest/response/ResponseStreamWriter.java
+3 -1
View File
@@ -12,6 +12,7 @@ import org.apache.commons.lang3.tuple.Triple;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.response.Response;
public interface ResponseStreamWriter {
@@ -21,7 +22,8 @@ public interface ResponseStreamWriter {
*
* @param results The results results
* @param out The OutputStream
* @param response The Response
* @throws Exception thrown if writing to the output stream fails
*/
public void respond(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results, OutputStream out) throws Exception;
public void respond(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results, OutputStream out, Response response) throws Exception;
}
src/main/java/ch/psi/daq/queryrest/response/csv/CSVHTTPResponse.java
+6 -5
View File
@@ -19,11 +19,10 @@ import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueries;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.backend.BackendQueryImpl;
import ch.psi.daq.domain.query.operation.AggregationType;
import ch.psi.daq.domain.query.operation.Compression;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.operation.ResponseFormat;
import ch.psi.daq.domain.query.response.ResponseFormat;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
@@ -43,7 +42,8 @@ public class CSVHTTPResponse extends AbstractHTTPResponse {
}
@Override
public void respond(ApplicationContext context, DAQQueries queries, HttpServletResponse httpResponse) throws Exception {
public void respond(ApplicationContext context, DAQQueries queries, HttpServletResponse httpResponse)
throws Exception {
OutputStream out = handleCompressionAndResponseHeaders(httpResponse, CONTENT_TYPE);
// do csv specific validations
@@ -59,7 +59,7 @@ public class CSVHTTPResponse extends AbstractHTTPResponse {
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> result =
queryManager.getEvents(queries);
// write the response back to the client using java 8 streams
streamWriter.respond(result, out);
streamWriter.respond(result, out, this);
} catch (Exception e) {
LOGGER.error("Failed to execute query '{}'.", queries, e);
throw e;
@@ -68,7 +68,8 @@ public class CSVHTTPResponse extends AbstractHTTPResponse {
protected void validateQueries(DAQQueries queries) {
for (DAQQueryElement query : queries) {
if (!(query.getAggregation() == null || AggregationType.value.equals(query.getAggregation().getAggregationType()))) {
if (!(query.getAggregation() == null || AggregationType.value.equals(query.getAggregation()
.getAggregationType()))) {
// We allow only no aggregation or value aggregation as
// extrema: nested structure and not clear how to map it to one line
// index: value is an array of Statistics whose size is not clear at initialization time
src/main/java/ch/psi/daq/queryrest/response/csv/CSVResponseStreamWriter.java
+44 -17
View File
@@ -28,18 +28,24 @@ import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.lang3.tuple.Triple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import ch.psi.daq.common.stream.StreamIterable;
import ch.psi.daq.common.stream.StreamMatcher;
import ch.psi.daq.common.stream.match.MapCreator;
import ch.psi.daq.common.stream.match.MapFiller;
import ch.psi.daq.common.stream.match.Padder;
import ch.psi.daq.common.stream.match.StreamMatcher;
import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.backend.BackendQueryImpl;
import ch.psi.daq.domain.query.backend.analyzer.BackendQueryAnalyzer;
import ch.psi.daq.domain.query.mapping.Mapping;
import ch.psi.daq.domain.query.mapping.IncompleteStrategy;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.response.Response;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
/**
@@ -49,41 +55,52 @@ import ch.psi.daq.queryrest.response.ResponseStreamWriter;
public class CSVResponseStreamWriter implements ResponseStreamWriter {
private static final Logger LOGGER = LoggerFactory.getLogger(CSVResponseStreamWriter.class);
public static final Mapping DEFAULT_MAPPING = new Mapping(IncompleteStrategy.FILL_NULL);
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 = "";
public static final String FIELDNAME_EXTREMA = "extrema";
private static final Function<Pair<ChannelName, DataEvent>, ChannelName> KEY_PROVIDER = (pair) -> pair.getKey();
private static final Function<DataEvent, ChannelName> KEY_PROVIDER = (event) -> new ChannelName(event.getChannel(), event.getBackend());
// 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 ToLongFunction<DataEvent> MATCHER_PROVIDER = (event) -> event.getGlobalMillis() / 10L;
@Resource
private ApplicationContext context;
@Resource
private Function<BackendQuery, BackendQueryAnalyzer> queryAnalizerFactory;
@Override
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
final OutputStream out) throws Exception {
final OutputStream out, final Response response) throws Exception {
if(results.size() > 1){
throw new IllegalStateException("CSV format does not allow for multiple queries.");
}
AtomicReference<Exception> exception = new AtomicReference<>();
final Map<ChannelName, Stream<Pair<ChannelName, DataEvent>>> streams = new LinkedHashMap<>(results.size());
final Map<ChannelName, Stream<DataEvent>> streams = new LinkedHashMap<>(results.size());
final List<String> header = new ArrayList<>();
final Collection<Pair<ChannelName, Function<DataEvent, String>>> accessors = new ArrayList<>();
final AtomicReference<DAQQueryElement> daqQueryRef = new AtomicReference<>();
final AtomicReference<BackendQuery> backendQueryRef = new AtomicReference<>();
// prepare the streams
/* get DataEvent stream of all sub-queries for later match */
results.forEach(entry -> {
final DAQQueryElement query = entry.getKey();
daqQueryRef.compareAndSet(null, query);
entry.getValue()
.sequential()
.forEach(triple -> {
backendQueryRef.compareAndSet(null, triple.getLeft());
if (triple.getRight() instanceof Stream) {
setupChannelColumns(query, triple.getLeft(), triple.getMiddle(), header, accessors);
final Stream<Pair<ChannelName, DataEvent>> eventStream = ((Stream<DataEvent>) triple.getRight())
.map(dataEvent -> Pair.of(triple.getMiddle(), dataEvent));
final Stream<DataEvent> eventStream = ((Stream<DataEvent>) triple.getRight());
streams.put(triple.getMiddle(), eventStream);
} else {
final String message = String.format("Expect a DataEvent Stream for '%s'.", triple.getMiddle());
@@ -92,10 +109,20 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
});
});
Mapping mapping = daqQueryRef.get().getMappingOrDefault(DEFAULT_MAPPING);
Padder<ChannelName, DataEvent> padder = mapping.getIncomplete().getPadder(context, backendQueryRef.get());
// 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();
StreamMatcher<ChannelName, DataEvent, Map<ChannelName, DataEvent>> streamMatcher =
new StreamMatcher<>(
KEY_PROVIDER,
MATCHER_PROVIDER,
new MapCreator<>(),
new MapFiller<>(),
null,
padder,
streams.values());
Iterator<Map<ChannelName, DataEvent>> streamsMatchIter = streamMatcher.iterator();
// prepare csv output
CSVFormat csvFormat = CSVFormat.EXCEL.withDelimiter(DELIMITER_CVS);
@@ -111,14 +138,14 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
csvFilePrinter.printRecord(header);
while (streamsMatchIter.hasNext()) {
final Map<ChannelName, Pair<ChannelName, DataEvent>> match = streamsMatchIter.next();
final Map<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());
DataEvent event = match.get(accessorPair.getKey());
if (event != null) {
return accessorPair.getValue().apply(event);
} else {
return EMPTY_VALUE;
}
src/main/java/ch/psi/daq/queryrest/response/json/JSONHTTPResponse.java
+22 -5
View File
@@ -1,6 +1,7 @@
package ch.psi.daq.queryrest.response.json;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.stream.Stream;
@@ -18,9 +19,10 @@ import ch.psi.daq.domain.query.DAQQueries;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.operation.Compression;
import ch.psi.daq.domain.query.operation.ResponseFormat;
import ch.psi.daq.domain.query.response.ResponseFormat;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
public class JSONHTTPResponse extends AbstractHTTPResponse {
private static final Logger LOGGER = LoggerFactory.getLogger(JSONHTTPResponse.class);
@@ -31,7 +33,7 @@ public class JSONHTTPResponse extends AbstractHTTPResponse {
public JSONHTTPResponse() {
super(ResponseFormat.JSON);
}
public JSONHTTPResponse(Compression compression) {
this();
setCompression(compression);
@@ -41,16 +43,31 @@ public class JSONHTTPResponse extends AbstractHTTPResponse {
public void respond(ApplicationContext context, DAQQueries queries, HttpServletResponse response) throws Exception {
OutputStream out = handleCompressionAndResponseHeaders(response, CONTENT_TYPE);
boolean hasMapping = false;
Iterator<DAQQueryElement> iter = queries.getQueries().iterator();
while (!hasMapping && iter.hasNext()) {
DAQQueryElement daqQueryElement = iter.next();
if (daqQueryElement.getMapping() != null) {
hasMapping = true;
}
}
try {
LOGGER.debug("Executing query '{}'", queries);
QueryManager queryManager = context.getBean(QueryManager.class);
JSONResponseStreamWriter streamWriter = context.getBean(JSONResponseStreamWriter.class);
ResponseStreamWriter streamWriter;
if (hasMapping) {
streamWriter = context.getBean(JSONTableResponseStreamWriter.class);
} else {
streamWriter = context.getBean(JSONResponseStreamWriter.class);
}
// execute query
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> result = queryManager.getEvents(queries);
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> result =
queryManager.getEvents(queries);
// write the response back to the client using java 8 streams
streamWriter.respond(result, out);
streamWriter.respond(result, out, this);
} catch (Exception e) {
LOGGER.error("Failed to execute query '{}'.", queries, e);
throw e;
src/main/java/ch/psi/daq/queryrest/response/json/JSONResponseStreamWriter.java
+13 -8
View File
@@ -29,6 +29,7 @@ import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.response.Response;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
/**
@@ -48,8 +49,8 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
private ObjectMapper mapper;
@Override
public void respond(List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
OutputStream out) throws Exception {
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
final OutputStream out, final Response response) throws Exception {
AtomicReference<Exception> exception = new AtomicReference<>();
JsonGenerator generator = jsonFactory.createGenerator(out, JsonEncoding.UTF8);
@@ -61,8 +62,8 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
results
.forEach(entryy -> {
DAQQueryElement daqQuery = entryy.getKey();
Set<String> includedFields = getFields(daqQuery);
ObjectWriter writer = configureWriter(includedFields);
Set<String> includedFields = getFields(daqQuery, true);
ObjectWriter writer = configureWriter(includedFields, mapper);
try {
generator.writeStartArray();
@@ -106,7 +107,7 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
}
}
protected Set<String> getFields(DAQQueryElement query) {
public static Set<String> getFields(DAQQueryElement query, boolean removeIdentifiers) {
Set<QueryField> queryFields = query.getFields();
List<Aggregation> aggregations = query.getAggregation() != null ? query.getAggregation().getAggregations() : null;
List<Extrema> extrema = query.getAggregation() != null ? query.getAggregation().getExtrema() : null;
@@ -129,8 +130,11 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
includedFields.add("extrema");
}
// do not write channel since it is already provided as key in mapping
includedFields.remove(QueryField.channel.name());
if (removeIdentifiers) {
// do not write channel/backend since it is already provided as key in mapping
includedFields.remove(QueryField.channel.name());
includedFields.remove(QueryField.backend.name());
}
return includedFields;
}
@@ -141,9 +145,10 @@ public class JSONResponseStreamWriter implements ResponseStreamWriter {
*
* @param includedFields set of strings which correspond to the getter method names of the
* classes registered as a mixed-in
* @param mapper The ObjectMapper
* @return the configured writer that includes the specified fields
*/
private ObjectWriter configureWriter(Set<String> includedFields) {
public static ObjectWriter configureWriter(Set<String> includedFields, ObjectMapper mapper) {
SimpleFilterProvider propertyFilter = new SimpleFilterProvider();
propertyFilter.addFilter("namedPropertyFilter", SimpleBeanPropertyFilter.filterOutAllExcept(includedFields));
// only write the properties not excluded in the filter
src/main/java/ch/psi/daq/queryrest/response/json/JSONTableResponseStreamWriter.java
+195
View File
@@ -0,0 +1,195 @@
package ch.psi.daq.queryrest.response.json;
import java.io.IOException;
import java.io.OutputStream;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
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.Collectors;
import java.util.stream.Stream;
import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import javax.servlet.ServletResponse;
import org.apache.commons.lang3.tuple.Triple;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import ch.psi.daq.common.stream.match.ListCreator;
import ch.psi.daq.common.stream.match.ListFiller;
import ch.psi.daq.common.stream.match.Padder;
import ch.psi.daq.common.stream.match.StreamMatcher;
import ch.psi.daq.common.time.TimeUtils;
import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.query.DAQQueryElement;
import ch.psi.daq.domain.query.backend.BackendQuery;
import ch.psi.daq.domain.query.bin.BinningStrategy;
import ch.psi.daq.domain.query.bin.strategy.BinningStrategyPerBinPulse;
import ch.psi.daq.domain.query.bin.strategy.BinningStrategyPerBinTime;
import ch.psi.daq.domain.query.mapping.Mapping;
import ch.psi.daq.domain.query.mapping.IncompleteStrategy;
import ch.psi.daq.domain.query.operation.Aggregation;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.domain.query.response.Response;
import ch.psi.daq.domain.request.range.RequestRange;
import ch.psi.daq.query.bin.aggregate.BinnedValueCombiner;
import ch.psi.daq.queryrest.config.QueryRestConfig;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
/**
* Takes a Java 8 stream and writes it to the output stream provided by the {@link ServletResponse}
* of the current request.
*/
public class JSONTableResponseStreamWriter implements ResponseStreamWriter {
private static final Logger LOGGER = LoggerFactory.getLogger(JSONTableResponseStreamWriter.class);
public static final Mapping DEFAULT_MAPPING = new Mapping(IncompleteStrategy.KEEP_AS_IS);
private static final long MILLIS_PER_PULSE = TimeUtils.MILLIS_PER_PULSE;
private static final Function<DataEvent, ChannelName> KEY_PROVIDER = (event) -> new ChannelName(event.getChannel(),
event.getBackend());
// try to match sync data (bsread) with non sync data (epics) based on the time usin 10 millis
// buckets.
private static final ToLongFunction<DataEvent> MATCHER_PROVIDER = (event) -> event.getGlobalMillis()
/ MILLIS_PER_PULSE;
@Resource
private ApplicationContext context;
@Resource
private JsonFactory jsonFactory;
@Resource
private ObjectMapper mapper;
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS)
private Set<Aggregation> defaultResponseAggregations;
private Set<String> defaultResponseAggregationsStr;
@PostConstruct
public void afterPropertiesSet() {
defaultResponseAggregationsStr =
defaultResponseAggregations.stream().map(Aggregation::name)
.collect(Collectors.toCollection(LinkedHashSet::new));
}
@Override
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
final OutputStream out, final Response response) throws Exception {
AtomicReference<Exception> exception = new AtomicReference<>();
JsonGenerator generator = jsonFactory.createGenerator(out, JsonEncoding.UTF8);
try {
if (results.size() > 1) {
generator.writeStartArray();
}
results
.forEach(entryy -> {
DAQQueryElement daqQuery = entryy.getKey();
Set<String> includedFields = JSONResponseStreamWriter.getFields(daqQuery, false);
/* make sure identifiers are available */
includedFields.add(QueryField.channel.name());
includedFields.add(QueryField.backend.name());
includedFields.addAll(defaultResponseAggregationsStr);
ObjectWriter writer = JSONResponseStreamWriter.configureWriter(includedFields, mapper);
/* get DataEvent stream of sub-queries for later match */
final Map<ChannelName, Stream<DataEvent>> streams =
new LinkedHashMap<>(results.size());
final AtomicReference<BackendQuery> backendQueryRef = new AtomicReference<>();
entryy.getValue()
.sequential()
.forEach(
triple -> {
backendQueryRef.compareAndSet(null, triple.getLeft());
if (triple.getRight() instanceof Stream) {
streams.put(triple.getMiddle(), ((Stream<DataEvent>) triple.getRight()));
} else {
final String message =
String.format("Expect a DataEvent Stream for '%s' but got '%s'.",
triple.getMiddle(), triple.getRight().getClass().getSimpleName());
LOGGER.warn(message);
streams.put(triple.getMiddle(), Stream.empty());
}
});
BackendQuery backendQuery = backendQueryRef.get();
RequestRange requestRange = backendQuery.getRequest().getRequestRange();
BinningStrategy binningStrategy = backendQuery.getBinningStrategy();
Mapping mapping = daqQuery.getMappingOrDefault(DEFAULT_MAPPING);
Padder<ChannelName, DataEvent> padder = mapping.getIncomplete().getPadder(context, backendQuery);
ToLongFunction<DataEvent> matchProvider = binningStrategy;
if (binningStrategy == null) {
matchProvider = MATCHER_PROVIDER;
if (requestRange.isPulseIdRangeDefined()) {
binningStrategy = new BinningStrategyPerBinPulse(1);
} else if (requestRange.isTimeRangeDefined()) {
binningStrategy = new BinningStrategyPerBinTime(MILLIS_PER_PULSE);
} else {
String message = "Either time or pulseId range must be defined by the query!";
LOGGER.error(message);
throw new IllegalStateException(message);
}
}
binningStrategy.setRequestRange(requestRange);
/* online matching of the stream's content */
StreamMatcher<ChannelName, DataEvent, List<DataEvent>> streamMatcher =
new StreamMatcher<>(
KEY_PROVIDER,
matchProvider,
new ListCreator<ChannelName, DataEvent>(),
new ListFiller<ChannelName, DataEvent>(),
new BinnedValueCombiner(binningStrategy),
padder,
streams.values());
Iterator<List<DataEvent>> streamsMatchIter = streamMatcher.iterator();
try {
writer.writeValue(generator, streamsMatchIter);
} catch (Exception e) {
LOGGER.error("Exception while writing json for '{}'", daqQuery.getChannels(), e);
exception.compareAndSet(null, e);
}
});
} catch (IOException e) {
LOGGER.error("Could not write JSON.", e);
exception.compareAndSet(null, e);
} finally {
if (results.size() > 1) {
generator.writeEndArray();
}
generator.flush();
generator.close();
}
if (exception.get() != null) {
throw exception.get();
}
}
}