Reformatting.
This commit is contained in:
Fabian Märki
@ -42,152 +42,145 @@ 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 {
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 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) {
|
||||
/**
|
||||
* 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)
|
||||
throws Throwable {
|
||||
return getChannels(new ChannelsRequest(channelName));
|
||||
}
|
||||
|
||||
ChannelsRequest request = new ChannelsRequest(channelName);
|
||||
Collection<String> specificChannels = getQueryProcessor(request.getDbMode()).getChannels(request.getRegex());
|
||||
return specificChannels;
|
||||
}
|
||||
/**
|
||||
* 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());
|
||||
|
||||
/**
|
||||
* 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());
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
public Stream<Entry<String, ?>> executeQuery(AbstractQuery query) {
|
||||
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
|
||||
|
||||
public Stream<Entry<String, ?>> executeQuery(AbstractQuery query) {
|
||||
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
|
||||
// all the magic happens here
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents = getQueryProcessor(query.getDbMode())
|
||||
.process(queryAnalizer);
|
||||
|
||||
// all the magic happens here
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents = getQueryProcessor(query.getDbMode())
|
||||
.process(queryAnalizer);
|
||||
// do post-process
|
||||
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
|
||||
|
||||
// do post-process
|
||||
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
|
||||
return channelToData;
|
||||
}
|
||||
|
||||
return channelToData;
|
||||
}
|
||||
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));
|
||||
}
|
||||
}
|
||||
|
||||
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));
|
||||
}
|
||||
}
|
||||
// ==========================================================================================
|
||||
// TODO: This is simply for initial / rudimentary testing - remove once
|
||||
// further evolved
|
||||
|
||||
// ==========================================================================================
|
||||
// TODO: This is simply for initial / rudimentary testing - remove once
|
||||
// further evolved
|
||||
@Resource
|
||||
private CassandraDataGen cassandraDataGen;
|
||||
|
||||
@Resource
|
||||
private CassandraDataGen cassandraDataGen;
|
||||
@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");
|
||||
|
||||
@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 -> 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 -> i, "TRFCB-channel3");
|
||||
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> nrOfElements - i,
|
||||
"TRFCB-channel4");
|
||||
}
|
||||
cassandraDataGen.writeData(3, 0, 4, i -> i, i -> 0, i -> i, i -> i, i -> 0, i -> nrOfElements - i,
|
||||
"TRFCB-channel4");
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user