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

Async read, query REST etc.

Browse Source
This commit is contained in:
Fabian Märki
2015-07-24 15:34:35 +02:00
parent 9b370cebe3
commit 689622be07
10 changed files with 282 additions and 221 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java
+59 -24
View File
@@ -3,7 +3,10 @@ package ch.psi.daq.queryrest.config;
import java.util.LinkedHashSet;
import java.util.Set;
import org.springframework.beans.factory.annotation.Autowired;
import javax.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
@@ -13,12 +16,17 @@ import org.springframework.util.StringUtils;
import com.fasterxml.jackson.annotation.JsonInclude.Include;
import com.fasterxml.jackson.core.JsonFactory;
import com.fasterxml.jackson.core.Version;
import com.fasterxml.jackson.databind.ObjectMapper;
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.query.config.QueryClientConfig;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.query.model.AbstractQuery;
import ch.psi.daq.query.model.QueryField;
import ch.psi.daq.queryrest.model.PropertyFilterMixin;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@@ -27,18 +35,44 @@ import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@PropertySource(value = {"file:${user.home}/.config/daq/queryrest.properties"}, ignoreResourceNotFound = true)
public class QueryRestConfig {
@Autowired
private Environment env;
// a nested configuration
// this guarantees that the ordering of the properties file is as expected
// see:
// https://jira.spring.io/browse/SPR-10409?focusedCommentId=101393&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-101393
@Configuration
@Import({QueryConfig.class, QueryClientConfig.class})
@Import({QueryConfig.class})
static class InnerConfiguration {
}
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestConfig.class);
public static final String BEAN_NAME_DEFAULT_RESPONSE_FIELDS = "defaultResponseFields";
@Resource
private Environment env;
@Bean
public ObjectMapper objectMapper() {
ObjectMapper mapper = new ObjectMapper();
String packageName = AbstractQuery.class.getPackage().getName();
Class<AbstractQuery> abstractQueryClass = AbstractQuery.class;
AttributeBasedDeserializer<AbstractQuery> abstractQueryDeserializer =
new AttributeBasedDeserializer<AbstractQuery>(abstractQueryClass).register(packageName);
SimpleModule module = new SimpleModule("PolymorphicAbstractQuery", Version.unknownVersion());
module.addDeserializer(abstractQueryClass, abstractQueryDeserializer);
mapper.registerModule(module);
// only include non-null values
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(StorelessStatistics.class, PropertyFilterMixin.class);
return mapper;
}
@Bean
public JsonFactory jsonFactory() {
return new JsonFactory();
@@ -49,27 +83,28 @@ public class QueryRestConfig {
return new ResponseStreamWriter();
}
@Bean
public ObjectMapper objectMapper() {
ObjectMapper mapper = new ObjectMapper();
// only include non-null values
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(StorelessStatistics.class, PropertyFilterMixin.class);
return mapper;
}
@Bean
public Set<String> defaultResponseFields() {
String[] responseFields = StringUtils.commaDelimitedListToStringArray(env.getProperty("queryrest.default.response.fields"));
@Bean(name = BEAN_NAME_DEFAULT_RESPONSE_FIELDS)
public Set<QueryField> defaultResponseFields() {
String[] responseFields =
StringUtils.commaDelimitedListToStringArray(env.getProperty("queryrest.default.response.fields"));
// preserve order
LinkedHashSet<String> defaultResponseFields = new LinkedHashSet<>(responseFields.length);
for (String field : defaultResponseFields) {
defaultResponseFields.add(field);
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);
}
}
return defaultResponseFields;
}
// ==========================================================================================
// TODO: This is simply for initial / rudimentary testing - remove once further evolved
@Bean
public CassandraDataGen cassandraDataGen() {
return new CassandraDataGen();
}
}
src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java
+75 -106
View File
@@ -1,145 +1,114 @@
package ch.psi.daq.queryrest.controller;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.LongFunction;
import java.util.function.LongUnaryOperator;
import java.util.stream.Collectors;
import java.util.stream.LongStream;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.Map.Entry;
import java.util.Set;
import java.util.stream.Stream;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.Assert;
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;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import ch.psi.daq.cassandra.writer.CassandraWriter;
import ch.psi.daq.domain.DataType;
import ch.psi.daq.domain.cassandra.ChannelEvent;
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
import ch.psi.daq.domain.cassandra.DataEvent;
import ch.psi.daq.query.model.AbstractQuery;
import ch.psi.daq.query.model.PulseRangeQuery;
import ch.psi.daq.query.model.TimeRangeQuery;
import ch.psi.daq.query.model.QueryField;
import ch.psi.daq.query.processor.QueryProcessor;
import ch.psi.daq.queryrest.config.QueryRestConfig;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
@RestController
public class QueryRestController {
private static final Logger logger = LoggerFactory.getLogger(QueryRestController.class);
@Autowired
private CassandraWriter cassandraWriter;
@Autowired
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestController.class);
public static final String CHANNELS = "channels";
public static final String CHANNELS_REGEX = CHANNELS + "/{regex}";
public static final String QUERY = "query";
@Resource
private ResponseStreamWriter responseStreamWriter;
@Autowired
@Resource
private QueryProcessor queryProcessor;
@RequestMapping(value = "/pulserange")
public void pulseRange(@RequestBody PulseRangeQuery query, HttpServletResponse res) throws IOException {
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_FIELDS)
private Set<QueryField> defaultResponseFields;
logger.debug("PulseRangeQuery received: {}", query);
executeQuery(query, res);
@RequestMapping(
value = CHANNELS,
method = RequestMethod.GET,
produces = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody Collection<String> getChannels() throws Throwable {
try {
return queryProcessor.getChannels();
} catch (Throwable t) {
LOGGER.error("Failed to query channel names.", t);
throw t;
}
}
@RequestMapping(value = "/timerange")
public void timeRange(@RequestBody TimeRangeQuery query, HttpServletResponse res) throws IOException {
logger.debug("TimeRangeQuery received: {}", query);
executeQuery(query, res);
@RequestMapping(
value = CHANNELS_REGEX,
method = RequestMethod.GET,
produces = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody Collection<String> getChannels(@PathVariable String regex) throws Throwable {
try {
return queryProcessor.getChannels(regex);
} catch (Throwable t) {
LOGGER.error("Failed to query channel names with regex '{}'.", regex, t);
throw t;
}
}
private void executeQuery(AbstractQuery query, HttpServletResponse res) throws IOException {
@RequestMapping(
value = QUERY,
method = RequestMethod.GET,
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());
// all the magic happens here
Map<String, Stream<? extends DataEvent>> process = queryProcessor.process(query);
validateQuery(query);
Stream<DataEvent> flatStreams = process.values().stream().flatMap(s -> {
return s;
});
// all the magic happens here
Stream<Entry<String, Stream<? extends DataEvent>>> process = queryProcessor.process(query);
// write the response back to the client using java 8 streams
responseStreamWriter.respond(flatStreams, query, res);
// write the response back to the client using java 8 streams
responseStreamWriter.respond(process, 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) {
if (query.getFields() == null || query.getFields().isEmpty()) {
query.setFields(new LinkedHashSet<>(defaultResponseFields));
}
}
// ==========================================================================================
// TODO this is simply a method for initial / rudimentary testing - remove once further evolved
// TODO: This is simply for initial / rudimentary testing - remove once further evolved
@Resource
private CassandraDataGen cassandraDataGen;
@RequestMapping(value = "/write")
public long writeDummyEntry() {
long startIndex = System.currentTimeMillis();
writeData(3,startIndex, 100, new String[]{"channel1", "channel2"});
return startIndex;
}
private void writeData(int dataReplication, long startIndex, long nrOfElements,
String... channelNames) {
writeData(dataReplication, startIndex, nrOfElements,
i -> 2 * i,
i -> 2 * i,
i -> 2 * i,
i -> 2 * i,
i -> 2 * i,
i -> Long.valueOf(2 * i),
channelNames);
}
private <T> void writeData(int dataReplication, long startIndex, long nrOfElements,
LongUnaryOperator iocMillis, LongUnaryOperator iocNanos, LongUnaryOperator pulseIds,
LongUnaryOperator globalMillis, LongUnaryOperator globalNanos, LongFunction<T> valueFunction,
String... channelNames) {
Assert.notNull(channelNames);
CompletableFuture<Void> future;
List<ChannelEvent> events =
Arrays.stream(channelNames)
.parallel()
.flatMap(
channelName -> {
Stream<ChannelEvent> stream =
LongStream
.range(startIndex, startIndex + nrOfElements)
.parallel()
.mapToObj(
i -> {
Object value = valueFunction.apply(i);
return new ChannelEvent(channelName,
iocMillis.applyAsLong(i),
iocNanos.applyAsLong(i),
pulseIds.applyAsLong(i),
globalMillis.applyAsLong(i),
globalNanos.applyAsLong(i),
value,
DataType.getTypeName(value.getClass())
);
}
);
return stream;
})
.collect(Collectors.toList());
future = cassandraWriter.writeAsync(dataReplication, (int) TimeUnit.HOURS.toSeconds(1), events);
future.join();
public void writeDummyEntry() {
cassandraDataGen.writeData(3, 0, 100, "channel1", "channel2");
}
}
src/main/java/ch/psi/daq/queryrest/model/PropertyFilterMixin.java
+1 -1
View File
@@ -3,7 +3,7 @@ package ch.psi.daq.queryrest.model;
import com.fasterxml.jackson.annotation.JsonFilter;
/**
*
* Kind of marker for ObjectMapper MixIn
*/
@JsonFilter("namedPropertyFilter")
public class PropertyFilterMixin {
src/main/java/ch/psi/daq/queryrest/response/ResponseStreamWriter.java
+49 -20
View File
@@ -2,14 +2,16 @@ package ch.psi.daq.queryrest.response;
import java.io.IOException;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Set;
import java.util.Map.Entry;
import java.util.stream.Stream;
import javax.annotation.Resource;
import javax.servlet.ServletResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import com.fasterxml.jackson.core.JsonEncoding;
import com.fasterxml.jackson.core.JsonFactory;
@@ -22,6 +24,7 @@ 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.QueryField;
/**
* Takes a Java 8 stream and writes it to the output stream provided by the {@link ServletResponse}
@@ -31,15 +34,12 @@ public class ResponseStreamWriter {
private static final Logger logger = LoggerFactory.getLogger(ResponseStreamWriter.class);
@Autowired
@Resource
private JsonFactory jsonFactory;
@Autowired
@Resource
private ObjectMapper mapper;
@Autowired
private Set<String> defaultResponseFields;
/**
* Responding with the the contents of the stream by writing into the output stream of the
* {@link ServletResponse}.
@@ -49,16 +49,26 @@ public class ResponseStreamWriter {
* @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<DataEvent> stream, AbstractQuery query, ServletResponse response) throws IOException {
public void respond(Stream<Entry<String, Stream<? extends DataEvent>>> stream, AbstractQuery query,
ServletResponse response) throws IOException {
Set<String> includedFields = query.getFieldsOrDefault(defaultResponseFields);
Set<QueryField> queryFields = query.getFields();
List<AggregationEnum> aggregations = query.getAggregations();
if (query.getAggregations() != null) {
includedFields = new LinkedHashSet<String>(includedFields);
Set<String> includedFields =
new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
for (QueryField field : queryFields) {
includedFields.add(field.name());
}
if (aggregations != null) {
for (AggregationEnum aggregation : query.getAggregations()) {
includedFields.add(aggregation.name());
}
}
// do not write channel since it is already provided as key in mapping
includedFields.remove(QueryField.channel.name());
ObjectWriter writer = configureWriter(includedFields);
respondInternal(stream, response, writer);
}
@@ -87,20 +97,39 @@ public class ResponseStreamWriter {
* @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<DataEvent> stream, ServletResponse response, ObjectWriter writer)
private void respondInternal(Stream<Entry<String, Stream<? extends DataEvent>>> stream, ServletResponse response,
ObjectWriter writer)
throws IOException {
JsonGenerator generator = jsonFactory.createGenerator(response.getOutputStream(), JsonEncoding.UTF8);
generator.writeStartArray();
stream.forEach(ds -> {
try {
logger.trace("Writing value for: {}", ds);
// use the writer created just before
writer.writeValue(generator, ds);
} catch (Exception e) {
logger.error("", e);
}
});
stream
/* ensure elements are sequentially written to the stream */
.sequential()
.forEach(
entry -> {
try {
generator.writeStartObject();
generator.writeStringField(QueryField.channel.name(), entry.getKey());
generator.writeArrayFieldStart("values");
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.writeEndObject();
} catch (Exception e) {
logger.error("Could not write channel name of channel '{}'", entry.getKey(), e);
}
});
generator.writeEndArray();
generator.flush();
generator.close();