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

ATEST-122

Browse Source
This commit is contained in:
Fabian Märki
2015-07-29 11:32:36 +02:00
parent 8ace7ec050
commit 8ce8370601
10 changed files with 278 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java
+46 -6
View File
@@ -1,7 +1,9 @@
package ch.psi.daq.queryrest.config;
import java.util.EnumMap;
import java.util.LinkedHashSet;
import java.util.Set;
import java.util.function.Function;
import javax.annotation.Resource;
@@ -23,11 +25,16 @@ import com.fasterxml.jackson.databind.module.SimpleModule;
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
import ch.psi.daq.common.json.deserialize.AttributeBasedDeserializer;
import ch.psi.daq.common.statistic.StorelessStatistics;
import ch.psi.daq.domain.cassandra.ChannelEvent;
import ch.psi.daq.domain.cassandra.DataEvent;
import ch.psi.daq.query.analyzer.CassandraQueryAnalyzer;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.query.model.AbstractQuery;
import ch.psi.daq.query.model.Aggregation;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.query.model.QueryField;
import ch.psi.daq.queryrest.model.PropertyFilterMixin;
import ch.psi.daq.queryrest.response.JsonStreamSerializer;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@Configuration
@@ -35,6 +42,10 @@ import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@PropertySource(value = {"file:${user.home}/.config/daq/queryrest.properties"}, ignoreResourceNotFound = true)
public class QueryRestConfig {
private static final String QUERYREST_DEFAULT_RESPONSE_AGGREGATIONS = "queryrest.default.response.aggregations";
private static final String QUERYREST_DEFAULT_RESPONSE_FIELDS = "queryrest.default.response.fields";
// a nested configuration
// this guarantees that the ordering of the properties file is as expected
// see:
@@ -47,6 +58,7 @@ public class QueryRestConfig {
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestConfig.class);
public static final String BEAN_NAME_DEFAULT_RESPONSE_FIELDS = "defaultResponseFields";
public static final String BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS = "defaultResponseAggregations";
@Resource
private Environment env;
@@ -67,8 +79,14 @@ public class QueryRestConfig {
mapper.setSerializationInclusion(Include.NON_NULL);
// Mixin which is used dynamically to filter out which properties get serialised and which
// won't. This way, the user can specify which columns are to be received.
mapper.addMixIn(ChannelEvent.class, PropertyFilterMixin.class);
mapper.addMixIn(DataEvent.class, PropertyFilterMixin.class);
mapper.addMixIn(StorelessStatistics.class, PropertyFilterMixin.class);
mapper.addMixIn(EnumMap.class, PropertyFilterMixin.class);
// defines how to writer inner Streams (i.e. Stream<Entry<String, Stream<?>>> toSerialize)
module = new SimpleModule("Streams API", Version.unknownVersion());
module.addSerializer(new JsonStreamSerializer());
mapper.registerModule(module);
return mapper;
}
@@ -78,6 +96,11 @@ public class QueryRestConfig {
return new JsonFactory();
}
@Bean
public Function<Query, QueryAnalyzer> queryAnalizerFactory() {
return (query) -> new CassandraQueryAnalyzer(query);
}
@Bean
public ResponseStreamWriter responseStreamWriter() {
return new ResponseStreamWriter();
@@ -86,21 +109,38 @@ public class QueryRestConfig {
@Bean(name = BEAN_NAME_DEFAULT_RESPONSE_FIELDS)
public Set<QueryField> defaultResponseFields() {
String[] responseFields =
StringUtils.commaDelimitedListToStringArray(env.getProperty("queryrest.default.response.fields"));
StringUtils.commaDelimitedListToStringArray(env.getProperty(QUERYREST_DEFAULT_RESPONSE_FIELDS));
// preserve order
LinkedHashSet<QueryField> defaultResponseFields = new LinkedHashSet<>(responseFields.length);
for (String field : responseFields) {
try {
defaultResponseFields.add(QueryField.valueOf(field));
} catch (Exception e) {
LOGGER.error("Field '{}' in queryrest.default.response.fields is invalid.", field, e);
LOGGER.error("Field '{}' in '{}' is invalid.", field, QUERYREST_DEFAULT_RESPONSE_FIELDS, e);
}
}
return defaultResponseFields;
}
@Bean(name = BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS)
public Set<Aggregation> defaultResponseAggregations() {
String[] responseAggregations =
StringUtils.commaDelimitedListToStringArray(env.getProperty(QUERYREST_DEFAULT_RESPONSE_AGGREGATIONS));
// preserve order
LinkedHashSet<Aggregation> defaultResponseAggregations = new LinkedHashSet<>(responseAggregations.length);
for (String aggregation : responseAggregations) {
try {
defaultResponseAggregations.add(Aggregation.valueOf(aggregation));
} catch (Exception e) {
LOGGER.error("Aggregation '{}' in '{}' is invalid.", aggregation, QUERYREST_DEFAULT_RESPONSE_AGGREGATIONS,
e);
}
}
return defaultResponseAggregations;
}
// ==========================================================================================
// TODO: This is simply for initial / rudimentary testing - remove once further evolved
@Bean
src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java
+28 -9
View File
@@ -3,8 +3,10 @@ package ch.psi.daq.queryrest.controller;
import java.io.IOException;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.LinkedList;
import java.util.Map.Entry;
import java.util.Set;
import java.util.function.Function;
import java.util.stream.Stream;
import javax.annotation.Resource;
@@ -13,7 +15,6 @@ import javax.servlet.http.HttpServletResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
@@ -22,7 +23,10 @@ import org.springframework.web.bind.annotation.RestController;
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
import ch.psi.daq.domain.cassandra.DataEvent;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.model.AbstractQuery;
import ch.psi.daq.query.model.Aggregation;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.query.model.QueryField;
import ch.psi.daq.query.processor.QueryProcessor;
import ch.psi.daq.queryrest.config.QueryRestConfig;
@@ -43,8 +47,14 @@ public class QueryRestController {
@Resource
private QueryProcessor queryProcessor;
@Resource
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_FIELDS)
private Set<QueryField> defaultResponseFields;
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS)
private Set<Aggregation> defaultResponseAggregations;
@RequestMapping(
value = CHANNELS,
@@ -61,9 +71,10 @@ public class QueryRestController {
@RequestMapping(
value = CHANNELS_REGEX,
method = RequestMethod.GET,
method = RequestMethod.POST,
consumes = {MediaType.APPLICATION_JSON_VALUE},
produces = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody Collection<String> getChannels(@PathVariable String regex) throws Throwable {
public @ResponseBody Collection<String> getChannels(@RequestBody String regex) throws Throwable {
try {
return queryProcessor.getChannels(regex);
} catch (Throwable t) {
@@ -74,31 +85,39 @@ public class QueryRestController {
@RequestMapping(
value = QUERY,
method = RequestMethod.GET,
method = RequestMethod.POST,
consumes = {MediaType.APPLICATION_JSON_VALUE},
produces = {MediaType.APPLICATION_JSON_VALUE})
public void executeQuery(@RequestBody AbstractQuery query, HttpServletResponse res) throws IOException {
try {
LOGGER.debug("Execute query '{}'", query.getClass().getSimpleName());
validateQuery(query);
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
queryAnalizer.validate();
extendQuery(query);
// all the magic happens here
Stream<Entry<String, Stream<? extends DataEvent>>> process = queryProcessor.process(query);
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents = queryProcessor.process(queryAnalizer);
// do post-process
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
// write the response back to the client using java 8 streams
responseStreamWriter.respond(process, query, res);
responseStreamWriter.respond(channelToData, query, res);
} catch (Throwable t) {
LOGGER.error("Failed execute query '{}'.", query, t);
throw t;
}
}
// TODO: can we do this with built-in validators?
private void validateQuery(AbstractQuery query) {
private void extendQuery(AbstractQuery query) {
if (query.getFields() == null || query.getFields().isEmpty()) {
query.setFields(new LinkedHashSet<>(defaultResponseFields));
}
if(query.getAggregations() == null || query.getAggregations().isEmpty()){
query.setAggregations(new LinkedList<>(defaultResponseAggregations));
}
}
// ==========================================================================================
src/main/java/ch/psi/daq/queryrest/response/JsonStreamSerializer.java
+24
View File
@@ -0,0 +1,24 @@
package ch.psi.daq.queryrest.response;
import java.io.IOException;
import java.util.Iterator;
import java.util.stream.Stream;
import com.fasterxml.jackson.core.JsonGenerationException;
import com.fasterxml.jackson.core.JsonGenerator;
import com.fasterxml.jackson.databind.SerializerProvider;
import com.fasterxml.jackson.databind.ser.std.StdSerializer;
public class JsonStreamSerializer extends StdSerializer<Stream<?>>{
private static final long serialVersionUID = 4695859735299703478L;
public JsonStreamSerializer() {
super(Stream.class, true);
}
@Override
public void serialize(Stream<?> stream, JsonGenerator jgen, SerializerProvider provider) throws IOException,
JsonGenerationException {
provider.findValueSerializer(Iterator.class, null).serialize(stream.iterator(), jgen, provider);
}
}
src/main/java/ch/psi/daq/queryrest/response/ResponseStreamWriter.java
+14 -25
View File
@@ -21,9 +21,8 @@ import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
import ch.psi.daq.domain.cassandra.DataEvent;
import ch.psi.daq.query.model.AbstractQuery;
import ch.psi.daq.query.model.AggregationEnum;
import ch.psi.daq.query.model.Aggregation;
import ch.psi.daq.query.model.QueryField;
/**
@@ -44,16 +43,16 @@ public class ResponseStreamWriter {
* Responding with the the contents of the stream by writing into the output stream of the
* {@link ServletResponse}.
*
* @param stream {@link Stream} instance of {@link DataEvent}s
* @param stream Mapping from channel name to data
* @param query concrete instance of {@link AbstractQuery}
* @param response {@link ServletResponse} instance given by the current HTTP request
* @throws IOException thrown if writing to the output stream fails
*/
public void respond(Stream<Entry<String, Stream<? extends DataEvent>>> stream, AbstractQuery query,
public void respond(Stream<Entry<String, ?>> stream, AbstractQuery query,
ServletResponse response) throws IOException {
Set<QueryField> queryFields = query.getFields();
List<AggregationEnum> aggregations = query.getAggregations();
List<Aggregation> aggregations = query.getAggregations();
Set<String> includedFields =
new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
@@ -62,7 +61,7 @@ public class ResponseStreamWriter {
includedFields.add(field.name());
}
if (aggregations != null) {
for (AggregationEnum aggregation : query.getAggregations()) {
for (Aggregation aggregation : query.getAggregations()) {
includedFields.add(aggregation.name());
}
}
@@ -88,43 +87,33 @@ public class ResponseStreamWriter {
ObjectWriter writer = mapper.writer(propertyFilter);
return writer;
}
/**
* Writes the Java stream into the output stream.
* Writes the outer Java stream into the output stream.
*
* @param stream {@link Stream} instance of {@link DataEvent}s
* @param stream Mapping from channel name to data
* @param response {@link ServletResponse} instance given by the current HTTP request
* @param writer configured writer that includes the fields the end user wants to see
* @throws IOException thrown if writing to the output stream fails
*/
private void respondInternal(Stream<Entry<String, Stream<? extends DataEvent>>> stream, ServletResponse response,
private void respondInternal(Stream<Entry<String, ?>> stream, ServletResponse response,
ObjectWriter writer)
throws IOException {
JsonGenerator generator = jsonFactory.createGenerator(response.getOutputStream(), JsonEncoding.UTF8);
generator.writeStartArray();
stream
/* ensure elements are sequentially written to the stream */
/* ensure elements are sequentially written */
.sequential()
.forEach(
entry -> {
try {
generator.writeStartObject();
generator.writeStringField(QueryField.channel.name(), entry.getKey());
generator.writeArrayFieldStart("data");
entry.getValue()
/* ensure elements are sequentially written to the stream */
.sequential()
.forEach(
dataEvent -> {
try {
writer.writeValue(generator, dataEvent);
} catch (Exception e) {
logger.error("Could not write event with pulse-id '{}' of channel '{}'",
dataEvent.getPulseId(), entry.getKey(), e);
}
});
generator.writeEndArray();
generator.writeFieldName("data");
writer.writeValue(generator, entry.getValue());
generator.writeEndObject();
} catch (Exception e) {
logger.error("Could not write channel name of channel '{}'", entry.getKey(), e);