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

ATEST-530

Browse Source
This commit is contained in:
Fabian Märki
2016-08-18 13:42:58 +02:00
parent 337a951d10
commit 5c90c3c30b
22 changed files with 737 additions and 749 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java
+16 -11
View File
@@ -29,14 +29,14 @@ import com.fasterxml.jackson.databind.ObjectMapper;
import ch.psi.daq.common.statistic.Statistics;
import ch.psi.daq.domain.DataEvent;
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.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.analyzer.QueryAnalyzerImpl;
import ch.psi.daq.query.analyzer.BackendQueryAnalyzerImpl;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.queryrest.controller.validator.QueryValidator;
import ch.psi.daq.queryrest.model.PropertyFilterMixin;
import ch.psi.daq.queryrest.query.QueryManager;
@@ -46,7 +46,7 @@ import ch.psi.daq.queryrest.response.csv.CSVResponseStreamWriter;
import ch.psi.daq.queryrest.response.json.JSONResponseStreamWriter;
@Configuration
@Import(value=QueryRestConfigCORS.class)
@Import(value = QueryRestConfigCORS.class)
@PropertySource(value = {"classpath:queryrest.properties"})
@PropertySource(value = {"file:${user.home}/.config/daq/queryrest.properties"}, ignoreResourceNotFound = true)
public class QueryRestConfig extends WebMvcConfigurerAdapter {
@@ -66,6 +66,11 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
static class InnerConfiguration {
}
// somehow needed to make sure @Import elements will get initialized and afterPropertiesSet will
// get called (ensuring BackendAcces gets initialized according hierarchy)
@Resource
private QueryConfig queryConfig;
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestConfig.class);
public static final String BEAN_NAME_DEFAULT_RESPONSE_FIELDS = "defaultResponseFields";
@@ -78,7 +83,7 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
@Resource
private ObjectMapper objectMapper;
@PostConstruct
public void afterPropertiesSet() {
// only include non-null values
@@ -89,10 +94,10 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
objectMapper.addMixIn(Statistics.class, PropertyFilterMixin.class);
objectMapper.addMixIn(AbstractExtremaMeta.class, PropertyFilterMixin.class);
objectMapper.addMixIn(EnumMap.class, PropertyFilterMixin.class);
objectMapper.addMixIn(Response.class, PolymorphicResponseMixIn.class);
}
/**
* {@inheritDoc}
@@ -115,8 +120,8 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
}
@Bean
public Function<Query, QueryAnalyzer> queryAnalizerFactory() {
return (query) -> new QueryAnalyzerImpl(query);
public Function<BackendQuery, BackendQueryAnalyzer> queryAnalizerFactory() {
return (query) -> new BackendQueryAnalyzerImpl(query);
}
@Bean
@@ -128,9 +133,9 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
public CSVResponseStreamWriter csvResponseStreamWriter() {
return new CSVResponseStreamWriter();
}
@Bean
public QueryManager queryManager(){
public QueryManager queryManager() {
return new QueryManagerImpl();
}
src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java
+23 -150
View File
@@ -1,16 +1,9 @@
package ch.psi.daq.queryrest.controller;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import javax.annotation.Resource;
import javax.servlet.http.HttpServletResponse;
import javax.validation.Valid;
@@ -30,10 +23,14 @@ import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
import ch.psi.daq.common.ordering.Ordering;
import ch.psi.daq.domain.backend.Backend;
import ch.psi.daq.domain.config.DomainConfig;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.json.status.channel.ChannelStatus;
import ch.psi.daq.domain.json.channels.info.ChannelInfos;
import ch.psi.daq.domain.query.ChannelNameRequest;
import ch.psi.daq.domain.query.DAQQueries;
import ch.psi.daq.domain.query.DAQQuery;
import ch.psi.daq.domain.query.channels.ChannelsRequest;
@@ -44,33 +41,17 @@ 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.status.channel.ChannelStatusQuery;
import ch.psi.daq.domain.reader.Backend;
import ch.psi.daq.domain.request.validate.RequestProviderValidator;
import ch.psi.daq.domain.status.StatusReader;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.query.processor.ChannelNameCache;
import ch.psi.daq.query.processor.QueryProcessor;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
import ch.psi.daq.queryrest.response.PolymorphicResponseMixIn;
import ch.psi.daq.queryrest.response.json.JSONHTTPResponse;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.Lists;
@RestController
public class QueryRestController {
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestController.class);
public static final String PATH_CHANNELS = DomainConfig.PATH_CHANNELS;
public static final String PATH_QUERY = DomainConfig.PATH_QUERY;
public static final String PATH_QUERIES = DomainConfig.PATH_QUERIES;
public static final String PATH_STATUS_CHANNELS = DomainConfig.PATH_STATUS_CHANNELS;
@Resource
private ApplicationContext appContext;
@@ -86,91 +67,8 @@ public class QueryRestController {
private Response defaultResponse = new JSONHTTPResponse();
@Resource
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
@Resource(name = DomainConfig.BEAN_NAME_READ_TIMEOUT)
private Integer readTimeout;
private Map<Backend, QueryProcessor> queryProcessors = new LinkedHashMap<>();
private ChannelNameCache channelNameCache;
private Map<Backend, StatusReader> statusReaders = new LinkedHashMap<>();
@PostConstruct
public void afterPropertiesSet() {
List<Exception> exceptions = new ArrayList<>();
try {
QueryProcessor queryProcessor =
appContext.getBean(QueryConfig.BEAN_NAME_CASSANDRA_QUERY_PROCESSOR, QueryProcessor.class);
queryProcessors.put(queryProcessor.getBackend(), queryProcessor);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load query processor for cassandra.");
LOGGER.warn("##########");
LOGGER.warn("");
}
try {
QueryProcessor queryProcessor =
appContext.getBean(QueryConfig.BEAN_NAME_ARCHIVER_APPLIANCE_QUERY_PROCESSOR, QueryProcessor.class);
queryProcessors.put(queryProcessor.getBackend(), queryProcessor);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load query processor for archiverappliance.");
LOGGER.warn("##########");
LOGGER.warn("");
}
if (queryProcessors.isEmpty()) {
LOGGER.error("No query processor could be loaded! Exceptions were: ");
for (Exception exception : exceptions) {
LOGGER.error("", exception);
}
throw new RuntimeException("No Backends available!");
}
try {
StatusReader statusReader =
appContext.getBean(QueryConfig.BEAN_NAME_CASSANDRA_STATUS_READER, StatusReader.class);
statusReaders.put(statusReader.getBackend(), statusReader);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load status reader for cassandra.");
LOGGER.warn("##########");
LOGGER.warn("");
}
try {
StatusReader statusReader =
appContext.getBean(QueryConfig.BEAN_NAME_ARCHIVER_APPLIANCE_STATUS_READER, StatusReader.class);
statusReaders.put(statusReader.getBackend(), statusReader);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load status reader for archiverappliance.");
LOGGER.warn("##########");
LOGGER.warn("");
}
channelNameCache =
new ChannelNameCache(queryProcessors, appContext.getBean(DomainConfig.BEAN_NAME_READ_TIMEOUT,
Integer.class).longValue());
}
@PreDestroy
public void destroy() {
channelNameCache.destroy();
}
public void afterPropertiesSet() {}
@InitBinder
protected void initBinder(WebDataBinder binder) {
@@ -184,7 +82,7 @@ public class QueryRestController {
}
}
@RequestMapping(value = PATH_CHANNELS, method = {RequestMethod.GET, RequestMethod.POST},
@RequestMapping(value = DomainConfig.PATH_CHANNELS, method = {RequestMethod.GET, RequestMethod.POST},
produces = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody List<ChannelsResponse> getChannels(@RequestBody(required = false) ChannelsRequest request)
throws Throwable {
@@ -198,7 +96,7 @@ public class QueryRestController {
* @return Collection of channel names matching the specified input channel name
* @throws Throwable in case something goes wrong
*/
@RequestMapping(value = PATH_CHANNELS + "/{channelName}", method = {RequestMethod.GET},
@RequestMapping(value = DomainConfig.PATH_CHANNELS + "/{channelName}", method = {RequestMethod.GET},
produces = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody Collection<ChannelsResponse> getChannels(@PathVariable(value = "channelName") String channelName)
throws Throwable {
@@ -206,44 +104,19 @@ public class QueryRestController {
}
/**
* Queries for channels status
* Queries for channels info
*
* @param query the {@link ChannelStatusQuery}
* @param res the {@link HttpServletResponse} instance associated with this request
* @return Collection of channel status info
* @param request the ChannelNameRequest
* @return Collection of ChannelInfos
* @throws Throwable in case something goes wrong
*/
@RequestMapping(
value = PATH_STATUS_CHANNELS,
value = DomainConfig.PATH_CHANNELS_INFO,
method = RequestMethod.POST,
consumes = {MediaType.APPLICATION_JSON_VALUE})
public @ResponseBody Collection<ChannelStatus> executeChannelStatusQuery(@RequestBody ChannelStatusQuery query,
HttpServletResponse res) throws Throwable {
// set backends if not defined yet
channelNameCache.setBackends(query.getChannels());
List<CompletableFuture<ChannelStatus>> futures = new ArrayList<>(query.getChannels().size());
for (ChannelName channel : query.getChannels()) {
StatusReader statusReader = statusReaders.get(channel.getBackend());
if (statusReader != null) {
futures.add(statusReader.getChannelStatusAsync(channel.getName()));
} else {
LOGGER.warn("There is no StatusReader available for '{}'.", channel.getBackend());
}
}
try {
List<ChannelStatus> retStatus = new ArrayList<>(futures.size());
for (CompletableFuture<ChannelStatus> completableFuture : futures) {
retStatus.add(completableFuture.get(readTimeout, TimeUnit.SECONDS));
}
return retStatus;
} catch (Exception e) {
String message = "Could not extract ChannelStatus within timelimits.";
LOGGER.error(message, e);
throw new IllegalStateException(message, e);
}
public @ResponseBody Collection<ChannelInfos> executeChannelInfoQuery(@RequestBody ChannelNameRequest request)
throws Throwable {
return queryManager.getChannelInfos(request);
}
/**
@@ -256,7 +129,7 @@ public class QueryRestController {
* {@link #executeQuery(DAQQuery, HttpServletResponse)} fails
*/
@RequestMapping(
value = PATH_QUERY,
value = DomainConfig.PATH_QUERY,
method = RequestMethod.GET)
public void executeQueryBodyAsString(@RequestParam String jsonBody, HttpServletResponse res) throws Exception {
DAQQuery query = objectMapper.readValue(jsonBody, DAQQuery.class);
@@ -271,7 +144,7 @@ public class QueryRestController {
* @throws Exception thrown if writing to the output stream fails
*/
@RequestMapping(
value = PATH_QUERY,
value = DomainConfig.PATH_QUERY,
method = RequestMethod.POST,
consumes = {MediaType.APPLICATION_JSON_VALUE})
public void executeQuery(@RequestBody @Valid DAQQuery query, HttpServletResponse res) throws Exception {
@@ -288,7 +161,7 @@ public class QueryRestController {
* {@link #executeQueries(DAQQueries, HttpServletResponse)} fails
*/
@RequestMapping(
value = PATH_QUERIES,
value = DomainConfig.PATH_QUERIES,
method = RequestMethod.GET)
public void executeQueriesBodyAsString(@RequestParam String jsonBody, HttpServletResponse res) throws Exception {
DAQQueries queries = objectMapper.readValue(jsonBody, DAQQueries.class);
@@ -299,8 +172,8 @@ public class QueryRestController {
* Catch-all query method for getting data from the backend for both JSON and CSV requests.
* <p>
* The {@link DAQQueries} object will contain the concrete subclass based on the combination of
* fields defined in the user's query. The AttributeBasedDeserializer decides which class
* to deserialize the information into and has been configured (see
* fields defined in the user's query. The AttributeBasedDeserializer decides which class to
* deserialize the information into and has been configured (see
* QueryRestConfig#afterPropertiesSet) accordingly.
*
* @param queries the {@link DAQQueries}
@@ -308,7 +181,7 @@ public class QueryRestController {
* @throws Exception thrown if writing to the output stream fails
*/
@RequestMapping(
value = PATH_QUERIES,
value = DomainConfig.PATH_QUERIES,
method = RequestMethod.POST,
consumes = {MediaType.APPLICATION_JSON_VALUE})
public void executeQueries(@RequestBody @Valid DAQQueries queries, HttpServletResponse res) throws Exception {
@@ -396,7 +269,7 @@ public class QueryRestController {
public @ResponseBody List<Backend> getDBModeValues() {
return Lists.newArrayList(Backend.values());
}
/**
* Returns the current list of {@link Compression}s available.
*
src/main/java/ch/psi/daq/queryrest/query/QueryManager.java
+9 -4
View File
@@ -1,5 +1,6 @@
package ch.psi.daq.queryrest.query;
import java.util.Collection;
import java.util.List;
import java.util.Map.Entry;
import java.util.stream.Stream;
@@ -7,16 +8,20 @@ import java.util.stream.Stream;
import org.apache.commons.lang3.tuple.Triple;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.json.channels.info.ChannelInfos;
import ch.psi.daq.domain.query.ChannelNameRequest;
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.channels.ChannelsRequest;
import ch.psi.daq.domain.query.channels.ChannelsResponse;
import ch.psi.daq.query.model.impl.BackendQuery;
public interface QueryManager {
List<ChannelsResponse> getChannels(ChannelsRequest request) throws Exception;
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> executeQueries(DAQQueries queries)
Collection<ChannelInfos> getChannelInfos(ChannelNameRequest request) throws Exception;
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> getEvents(DAQQueries queries)
throws Exception;
}
}
src/main/java/ch/psi/daq/queryrest/query/QueryManagerImpl.java
+67 -84
View File
@@ -1,9 +1,8 @@
package ch.psi.daq.queryrest.query;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -19,83 +18,45 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;
import ch.psi.daq.cassandra.config.CassandraConfig;
import ch.psi.daq.domain.DataEvent;
import ch.psi.daq.domain.backend.BackendAccess;
import ch.psi.daq.domain.config.DomainConfig;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.json.channels.info.ChannelInfos;
import ch.psi.daq.domain.query.ChannelNameRequest;
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.backend.analyzer.BackendQueryAnalyzer;
import ch.psi.daq.domain.query.channels.ChannelNameCache;
import ch.psi.daq.domain.query.channels.ChannelsRequest;
import ch.psi.daq.domain.query.channels.ChannelsResponse;
import ch.psi.daq.domain.reader.Backend;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.config.QueryConfig;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.query.model.impl.BackendQuery;
import ch.psi.daq.query.processor.ChannelNameCache;
import ch.psi.daq.query.processor.QueryProcessor;
import ch.psi.daq.domain.query.processor.QueryProcessor;
import ch.psi.daq.queryrest.query.model.ChannelInfosStreamImpl;
public class QueryManagerImpl implements QueryManager {
@SuppressWarnings("unused")
private static final Logger LOGGER = LoggerFactory.getLogger(QueryManagerImpl.class);
@Resource
private ApplicationContext appContext;
@Resource
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
private Map<Backend, QueryProcessor> queryProcessors = new LinkedHashMap<>();
private Function<BackendQuery, BackendQueryAnalyzer> queryAnalizerFactory;
@Resource(name = DomainConfig.BEAN_NAME_BACKEND_ACCESS)
private BackendAccess backendAccess;
@Resource(name = DomainConfig.BEAN_NAME_CHANNEL_NAME_CACHE)
private ChannelNameCache channelNameCache;
@PostConstruct
public void afterPropertiesSet() {
List<Exception> exceptions = new ArrayList<>();
public void afterPropertiesSet() {}
try {
QueryProcessor queryProcessor =
appContext.getBean(QueryConfig.BEAN_NAME_CASSANDRA_QUERY_PROCESSOR, QueryProcessor.class);
queryProcessors.put(queryProcessor.getBackend(), queryProcessor);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load query processor for cassandra.");
LOGGER.warn("##########");
LOGGER.warn("");
}
try {
QueryProcessor queryProcessor =
appContext.getBean(QueryConfig.BEAN_NAME_ARCHIVER_APPLIANCE_QUERY_PROCESSOR, QueryProcessor.class);
queryProcessors.put(queryProcessor.getBackend(), queryProcessor);
} catch (Exception e) {
exceptions.add(e);
LOGGER.warn("");
LOGGER.warn("##########");
LOGGER.warn("Could not load query processor for archiverappliance.");
LOGGER.warn("##########");
LOGGER.warn("");
}
if (queryProcessors.isEmpty()) {
LOGGER.error("No query processor could be loaded! Exceptions were: ");
for (Exception exception : exceptions) {
LOGGER.error("", exception);
}
throw new RuntimeException("No Backends available!");
}
channelNameCache =
new ChannelNameCache(queryProcessors, appContext.getBean(DomainConfig.BEAN_NAME_READ_TIMEOUT,
Integer.class).longValue());
}
@PreDestroy
public void destroy() {
channelNameCache.destroy();
}
public void destroy() {}
@Override
public List<ChannelsResponse> getChannels(ChannelsRequest request) {
// in case not specified use defaults (e.g. GET)
@@ -105,9 +66,33 @@ public class QueryManagerImpl implements QueryManager {
return channelNameCache.getChannels(request);
}
public Collection<ChannelInfos> getChannelInfos(ChannelNameRequest request) {
// set backends if not defined yet
channelNameCache.setBackends(request.getChannels());
Stream<ChannelInfos> stream = request.getRequestsByBackend().entrySet().stream()
.filter(entry ->
backendAccess.hasDataReader(entry.getKey())
&& backendAccess.hasChannelInfoReader(entry.getKey()))
.flatMap(entry -> {
return entry.getValue().getChannelInfos(entry.getKey(), backendAccess)
.entrySet().stream()
.map(innerEntry -> {
return new ChannelInfosStreamImpl(
new ChannelName(innerEntry.getKey(), entry.getKey()),
innerEntry.getValue()
);
}
);
});
// materialize
return stream.collect(Collectors.toList());
}
@Override
public List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> executeQueries(DAQQueries queries) {
public List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> getEvents(DAQQueries queries) {
// set backends if not defined yet
channelNameCache.setBackends(queries);
@@ -116,32 +101,30 @@ public class QueryManagerImpl implements QueryManager {
for (DAQQueryElement queryElement : queries) {
Stream<Triple<BackendQuery, ChannelName, ?>> resultStreams =
BackendQuery
BackendQueryImpl
.getBackendQueries(queryElement)
.stream()
.filter(query -> {
QueryProcessor processor = queryProcessors.get(query.getBackend());
if (processor != null) {
return true;
} else {
LOGGER.warn("There is no QueryProcessor available for '{}'", query.getBackend());
return false;
}
})
.flatMap(query -> {
QueryProcessor processor = queryProcessors.get(query.getBackend());
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
.filter(
query ->
backendAccess.hasDataReader(query.getBackend())
&& backendAccess.hasQueryProcessor(query.getBackend())
)
.flatMap(
query -> {
QueryProcessor processor = backendAccess.getQueryProcessor(query.getBackend());
BackendQueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
// all the magic happens here
Stream<Entry<ChannelName, Stream<? extends DataEvent>>> channelToDataEvents =
processor.process(queryAnalizer);
// do post-process
Stream<Entry<ChannelName, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
/* all the magic happens here */
Stream<Entry<ChannelName, Stream<? extends DataEvent>>> channelToDataEvents =
processor.process(queryAnalizer);
/* do post-process */
Stream<Entry<ChannelName, ?>> channelToData =
queryAnalizer.postProcess(channelToDataEvents);
return channelToData.map(entry -> {
return Triple.of(query, entry.getKey(), entry.getValue());
return channelToData.map(entry -> {
return Triple.of(query, entry.getKey(), entry.getValue());
});
});
});
// Now we have a stream that loads elements sequential BackendQuery by BackendQuery.
// By materializing the outer Stream the elements of all BackendQuery are loaded async
src/main/java/ch/psi/daq/queryrest/query/model/ChannelInfosStreamImpl.java
+44
View File
@@ -0,0 +1,44 @@
package ch.psi.daq.queryrest.query.model;
import java.util.Iterator;
import java.util.stream.Stream;
import com.fasterxml.jackson.annotation.JsonIgnore;
import ch.psi.daq.domain.json.ChannelName;
import ch.psi.daq.domain.json.channels.info.ChannelInfo;
import ch.psi.daq.domain.json.channels.info.ChannelInfos;
public class ChannelInfosStreamImpl implements ChannelInfos {
private ChannelName channel;
private Stream<? extends ChannelInfo> infos;
public ChannelInfosStreamImpl() {}
public ChannelInfosStreamImpl(ChannelName channel, Stream<? extends ChannelInfo> infos) {
this.channel = channel;
this.infos = infos;
}
@Override
public ChannelName getChannel() {
return channel;
}
public Stream<? extends ChannelInfo> getInfos() {
// can only be consumed once
return infos;
}
@JsonIgnore
@Override
public Iterator<ChannelInfo> iterator() {
return getChannelInfos().iterator();
}
@JsonIgnore
@Override
public Stream<ChannelInfo> getChannelInfos() {
return infos.map(info -> (ChannelInfo) info);
}
}
src/main/java/ch/psi/daq/queryrest/response/ResponseStreamWriter.java
+1 -1
View File
@@ -11,7 +11,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.query.model.impl.BackendQuery;
import ch.psi.daq.domain.query.backend.BackendQuery;
public interface ResponseStreamWriter {
src/main/java/ch/psi/daq/queryrest/response/csv/CSVHTTPResponse.java
+3 -2
View File
@@ -18,11 +18,12 @@ import ch.psi.daq.domain.FieldNames;
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.query.model.impl.BackendQuery;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
@@ -56,7 +57,7 @@ public class CSVHTTPResponse extends AbstractHTTPResponse {
// execute query
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> result =
queryManager.executeQueries(queries);
queryManager.getEvents(queries);
// write the response back to the client using java 8 streams
streamWriter.respond(result, out);
} catch (Exception e) {
src/main/java/ch/psi/daq/queryrest/response/csv/CSVResponseStreamWriter.java
+5 -5
View File
@@ -34,12 +34,12 @@ import ch.psi.daq.common.stream.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.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.query.analyzer.QueryAnalyzer;
import ch.psi.daq.query.model.Query;
import ch.psi.daq.query.model.impl.BackendQuery;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
/**
@@ -61,7 +61,7 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
.getGlobalMillis() / 10L;
@Resource
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
private Function<BackendQuery, BackendQueryAnalyzer> queryAnalizerFactory;
@Override
public void respond(final List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> results,
@@ -158,7 +158,7 @@ public class CSVResponseStreamWriter implements ResponseStreamWriter {
daqQuery.getAggregation() != null ? daqQuery.getAggregation().getAggregations() : null;
List<Extrema> extrema = daqQuery.getAggregation() != null ? daqQuery.getAggregation().getExtrema() : null;
QueryAnalyzer queryAnalyzer = queryAnalizerFactory.apply(backendQuery);
BackendQueryAnalyzer queryAnalyzer = queryAnalizerFactory.apply(backendQuery);
for (QueryField field : queryFields) {
if (!(QueryField.value.equals(field) && queryAnalyzer.isAggregationEnabled())) {
src/main/java/ch/psi/daq/queryrest/response/json/JSONHTTPResponse.java
+2 -2
View File
@@ -16,9 +16,9 @@ import org.springframework.http.MediaType;
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.operation.Compression;
import ch.psi.daq.domain.query.operation.ResponseFormat;
import ch.psi.daq.query.model.impl.BackendQuery;
import ch.psi.daq.queryrest.query.QueryManager;
import ch.psi.daq.queryrest.response.AbstractHTTPResponse;
@@ -48,7 +48,7 @@ public class JSONHTTPResponse extends AbstractHTTPResponse {
JSONResponseStreamWriter streamWriter = context.getBean(JSONResponseStreamWriter.class);
// execute query
List<Entry<DAQQueryElement, Stream<Triple<BackendQuery, ChannelName, ?>>>> result = queryManager.executeQueries(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);
} catch (Exception e) {
src/main/java/ch/psi/daq/queryrest/response/json/JSONResponseStreamWriter.java
+1 -1
View File
@@ -25,10 +25,10 @@ import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
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.operation.Aggregation;
import ch.psi.daq.domain.query.operation.Extrema;
import ch.psi.daq.domain.query.operation.QueryField;
import ch.psi.daq.query.model.impl.BackendQuery;
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
/**