ATEST-249:
- provide a channel search method with with the user can define all or part of the channel name
This commit is contained in:
@ -17,6 +17,7 @@ import org.springframework.http.MediaType;
|
||||
import org.springframework.validation.Validator;
|
||||
import org.springframework.web.bind.WebDataBinder;
|
||||
import org.springframework.web.bind.annotation.InitBinder;
|
||||
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;
|
||||
@ -25,6 +26,7 @@ import org.springframework.web.bind.annotation.RestController;
|
||||
|
||||
import ch.psi.daq.cassandra.request.validate.RequestProviderValidator;
|
||||
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
|
||||
import ch.psi.daq.common.json.deserialize.AttributeBasedDeserializer;
|
||||
import ch.psi.daq.domain.DataEvent;
|
||||
import ch.psi.daq.query.analyzer.QueryAnalyzer;
|
||||
import ch.psi.daq.query.model.Aggregation;
|
||||
@ -40,148 +42,152 @@ import ch.psi.daq.queryrest.response.ResponseStreamWriter;
|
||||
@RestController
|
||||
public class QueryRestController {
|
||||
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestController.class);
|
||||
private static final Logger LOGGER = LoggerFactory.getLogger(QueryRestController.class);
|
||||
|
||||
public static final String CHANNELS = "/channels";
|
||||
public static final String QUERY = "/query";
|
||||
public static final String CHANNELS = "/channels";
|
||||
public static final String QUERY = "/query";
|
||||
|
||||
@Resource
|
||||
private Validator queryValidator;
|
||||
private Validator requestProviderValidator = new RequestProviderValidator();
|
||||
@Resource
|
||||
private Validator queryValidator;
|
||||
private Validator requestProviderValidator = new RequestProviderValidator();
|
||||
|
||||
@Resource
|
||||
private ResponseStreamWriter responseStreamWriter;
|
||||
@Resource
|
||||
private ResponseStreamWriter responseStreamWriter;
|
||||
|
||||
@Resource
|
||||
private QueryProcessor cassandraQueryProcessor;
|
||||
@Resource
|
||||
private QueryProcessor cassandraQueryProcessor;
|
||||
|
||||
@Resource
|
||||
private QueryProcessor archiverApplianceQueryProcessor;
|
||||
@Resource
|
||||
private QueryProcessor archiverApplianceQueryProcessor;
|
||||
|
||||
@Resource
|
||||
private Function<Query, QueryAnalyzer> queryAnalizerFactory;
|
||||
@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_FIELDS)
|
||||
private Set<QueryField> defaultResponseFields;
|
||||
|
||||
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS)
|
||||
private Set<Aggregation> defaultResponseAggregations;
|
||||
@Resource(name = QueryRestConfig.BEAN_NAME_DEFAULT_RESPONSE_AGGREGATIONS)
|
||||
private Set<Aggregation> defaultResponseAggregations;
|
||||
|
||||
@InitBinder
|
||||
protected void initBinder(WebDataBinder binder) {
|
||||
if (binder.getTarget() != null) {
|
||||
if (requestProviderValidator.supports(binder.getTarget().getClass())) {
|
||||
binder.addValidators(requestProviderValidator);
|
||||
}
|
||||
if (queryValidator.supports(binder.getTarget().getClass())) {
|
||||
binder.addValidators(queryValidator);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@InitBinder
|
||||
protected void initBinder(WebDataBinder binder) {
|
||||
if (binder.getTarget() != null) {
|
||||
if (requestProviderValidator.supports(binder.getTarget().getClass())) {
|
||||
binder.addValidators(requestProviderValidator);
|
||||
}
|
||||
if (queryValidator.supports(binder.getTarget().getClass())) {
|
||||
binder.addValidators(queryValidator);
|
||||
}
|
||||
}
|
||||
}
|
||||
@RequestMapping(value = CHANNELS, method = { RequestMethod.GET, RequestMethod.POST }, produces = { MediaType.APPLICATION_JSON_VALUE })
|
||||
public @ResponseBody Collection<String> getChannels(@RequestBody(required = false) ChannelsRequest request)
|
||||
throws Throwable {
|
||||
// in case not specified use default (e.g. GET)
|
||||
if (request == null) {
|
||||
request = new ChannelsRequest();
|
||||
}
|
||||
|
||||
@RequestMapping(
|
||||
value = CHANNELS,
|
||||
method = {RequestMethod.GET, RequestMethod.POST},
|
||||
produces = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public @ResponseBody Collection<String> getChannels(@RequestBody(required = false) ChannelsRequest request)
|
||||
throws Throwable {
|
||||
// in case not specified use default (e.g. GET)
|
||||
if (request == null) {
|
||||
request = new ChannelsRequest();
|
||||
}
|
||||
try {
|
||||
// sorted collection
|
||||
Collection<String> allChannels = getQueryProcessor(request.getDbMode()).getChannels(request.getRegex());
|
||||
return allChannels;
|
||||
} catch (Throwable t) {
|
||||
LOGGER.error("Failed to query channel names.", t);
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
|
||||
try {
|
||||
return getQueryProcessor(request.getDbMode()).getChannels(request.getRegex());
|
||||
} catch (Throwable t) {
|
||||
LOGGER.error("Failed to query channel names.", t);
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
/**
|
||||
* Query specific channel names, and return only those.
|
||||
*
|
||||
* @param channelName
|
||||
* part of (or full) channel name
|
||||
* @return Collection of channel names matching the specified input channel
|
||||
* name
|
||||
*/
|
||||
@RequestMapping(value = CHANNELS + "/{channelName}", method = { RequestMethod.GET }, produces = { MediaType.APPLICATION_JSON_VALUE })
|
||||
public @ResponseBody Collection<String> getChannels(@PathVariable(value = "channelName") String channelName) {
|
||||
|
||||
@RequestMapping(
|
||||
value = QUERY,
|
||||
method = RequestMethod.POST,
|
||||
consumes = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public void executeQuery(@RequestBody @Valid AbstractQuery query, HttpServletResponse res) throws IOException {
|
||||
try {
|
||||
LOGGER.debug("Executing query '{}'", query.getClass().getSimpleName());
|
||||
ChannelsRequest request = new ChannelsRequest(channelName);
|
||||
Collection<String> specificChannels = getQueryProcessor(request.getDbMode()).getChannels(request.getRegex());
|
||||
return specificChannels;
|
||||
}
|
||||
|
||||
// write the response back to the client using java 8 streams
|
||||
responseStreamWriter.respond(executeQuery(query), query, res);
|
||||
} catch (Throwable t) {
|
||||
LOGGER.error("Failed execute query '{}'.", query, t);
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
|
||||
public Stream<Entry<String, ?>> executeQuery(AbstractQuery query) {
|
||||
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
|
||||
/**
|
||||
* Catch-all query method for getting data from the backend.
|
||||
* <p>
|
||||
* The {@link AbstractQuery} object will be a concrete subclass based on the
|
||||
* combination of fields defined in the user's query. The
|
||||
* {@link AttributeBasedDeserializer} decides which class to deserialize the
|
||||
* information into and has been configured (see
|
||||
* QueryRestConfig#afterPropertiesSet) accordingly.
|
||||
*
|
||||
* @param query
|
||||
* concrete implementation of {@link AbstractQuery}
|
||||
* @param res
|
||||
* the {@link HttpServletResponse} instance associated with this
|
||||
* request
|
||||
* @throws IOException
|
||||
* thrown if writing to the output stream fails
|
||||
*/
|
||||
@RequestMapping(value = QUERY, method = RequestMethod.POST, consumes = { MediaType.APPLICATION_JSON_VALUE })
|
||||
public void executeQuery(@RequestBody @Valid AbstractQuery query, HttpServletResponse res) throws IOException {
|
||||
try {
|
||||
LOGGER.debug("Executing query '{}'", query.toString());
|
||||
|
||||
// all the magic happens here
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents =
|
||||
getQueryProcessor(query.getDbMode()).process(queryAnalizer);
|
||||
// write the response back to the client using java 8 streams
|
||||
responseStreamWriter.respond(executeQuery(query), query, res);
|
||||
} catch (IOException t) {
|
||||
LOGGER.error("Failed to execute query '{}'.", query, t);
|
||||
throw t;
|
||||
}
|
||||
}
|
||||
|
||||
// do post-process
|
||||
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
|
||||
|
||||
return channelToData;
|
||||
}
|
||||
public Stream<Entry<String, ?>> executeQuery(AbstractQuery query) {
|
||||
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
|
||||
|
||||
private QueryProcessor getQueryProcessor(DBMode dbMode) {
|
||||
if (DBMode.databuffer.equals(dbMode)) {
|
||||
return cassandraQueryProcessor;
|
||||
} else if (DBMode.archiverappliance.equals(dbMode)) {
|
||||
return archiverApplianceQueryProcessor;
|
||||
} else {
|
||||
LOGGER.error("Unknown DBMode '{}'!", dbMode);
|
||||
throw new IllegalArgumentException(String.format("Unknown DBMode '%s'", dbMode));
|
||||
}
|
||||
}
|
||||
// all the magic happens here
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents = getQueryProcessor(query.getDbMode())
|
||||
.process(queryAnalizer);
|
||||
|
||||
// ==========================================================================================
|
||||
// TODO: This is simply for initial / rudimentary testing - remove once further evolved
|
||||
// do post-process
|
||||
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
|
||||
|
||||
@Resource
|
||||
private CassandraDataGen cassandraDataGen;
|
||||
return channelToData;
|
||||
}
|
||||
|
||||
@RequestMapping(value = "/write")
|
||||
public void writeDummyEntry() {
|
||||
long nrOfElements = 4;
|
||||
cassandraDataGen.writeData(3, 0, 4,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> new long[] {i, i, i, i},
|
||||
"TRFCA-channel1");
|
||||
private QueryProcessor getQueryProcessor(DBMode dbMode) {
|
||||
if (DBMode.databuffer.equals(dbMode)) {
|
||||
return cassandraQueryProcessor;
|
||||
} else if (DBMode.archiverappliance.equals(dbMode)) {
|
||||
return archiverApplianceQueryProcessor;
|
||||
} else {
|
||||
LOGGER.error("Unknown DBMode '{}'!", dbMode);
|
||||
throw new IllegalArgumentException(String.format("Unknown DBMode '%s'", dbMode));
|
||||
}
|
||||
}
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> new long[] {nrOfElements - i, nrOfElements - i, nrOfElements - i, nrOfElements - i},
|
||||
"TRFCA-channel2");
|
||||
// ==========================================================================================
|
||||
// TODO: This is simply for initial / rudimentary testing - remove once
|
||||
// further evolved
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> i,
|
||||
"TRFCB-channel3");
|
||||
@Resource
|
||||
private CassandraDataGen cassandraDataGen;
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> 0,
|
||||
i -> nrOfElements - i,
|
||||
"TRFCB-channel4");
|
||||
}
|
||||
@RequestMapping(value = "/write")
|
||||
public void writeDummyEntry() {
|
||||
long nrOfElements = 4;
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> new long[] { i, i, i, i },
|
||||
"TRFCA-channel1");
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> new long[] { nrOfElements - i,
|
||||
nrOfElements - i, nrOfElements - i, nrOfElements - i }, "TRFCA-channel2");
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> i, "TRFCB-channel3");
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> nrOfElements - i,
|
||||
"TRFCB-channel4");
|
||||
}
|
||||
}
|
||||
|
@ -14,6 +14,10 @@ public class ChannelsRequest {
|
||||
|
||||
public ChannelsRequest() {}
|
||||
|
||||
public ChannelsRequest(String regex) {
|
||||
this(DBMode.databuffer, regex);
|
||||
}
|
||||
|
||||
public ChannelsRequest(DBMode dbMode, String regex) {
|
||||
this.regex = regex;
|
||||
this.dbMode = dbMode;
|
||||
|
@ -74,3 +74,9 @@ curl -v -X POST -H 'Content-Type: application/json' -d '{"queryType":"timerange"
|
||||
|
||||
|
||||
curl -v -X POST -H 'Content-Type: application/json' -d '{"queryType":"pulserange","ordering":"ASC","channels":["dummy-test"],"fields":["channel","pulseId","globalMillis","globalNanos","dbValueBytes"],"binningStrategy":"bincount","binDuration":100,"aggregateChannels":"false","aggregationType":"index","aggregations":[{"fieldRef":"e_val","type":"max","resultFieldName":"maximum"},{"fieldRef":"e_val","type":"min","resultFieldName":"minimum"}], "queryRange":{"startPulseId":100,"endPulseId":100}}' http://localhost:8080/pulserange
|
||||
|
||||
|
||||
===============================================================================================================================================
|
||||
|
||||
|
||||
curl -H "Content-Type: application/json" -X POST -d '{"channels":["BooleanScalar"],"startPulseId":144404537,"endPulseId":144404547}' http://sf-nube-14.psi.ch:8080/query
|
Reference in New Issue
Block a user