Merge pull request #14 in ST/ch.psi.daq.queryrest from atest259 to master
# By Zellweger Christof Ralf # Via Zellweger Christof Ralf * commit 'fe8277df5e18007643fe6fb2b1bbfca68e8bfe84': ATEST-259: - clean up, improving naming ATEST-259: - removing unused test dependencies ATEST-259: - adding tests, re-implementing a simpler CORS filter than previously used ATEST-259: - adding simple CORS filter which includes some standard cors headers ATEST-259: - fixing QueryRestController tests before adding the CORS filter ATEST-259: - trying to implement dummy cassandra reader - fixing some of the tests
This commit is contained in:
Märki Fabian
@ -1,5 +1,5 @@
|
||||
#
|
||||
#Wed Sep 16 07:23:26 CEST 2015
|
||||
#Wed Oct 28 12:53:12 CET 2015
|
||||
org.eclipse.jdt.core.compiler.debug.localVariable=generate
|
||||
org.eclipse.jdt.core.compiler.compliance=1.8
|
||||
org.eclipse.jdt.core.compiler.codegen.unusedLocal=preserve
|
||||
|
||||
@ -9,9 +9,11 @@ import java.util.function.Function;
|
||||
|
||||
import javax.annotation.PostConstruct;
|
||||
import javax.annotation.Resource;
|
||||
import javax.servlet.Filter;
|
||||
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
@ -23,12 +25,6 @@ import org.springframework.util.StringUtils;
|
||||
import org.springframework.validation.Validator;
|
||||
import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter;
|
||||
|
||||
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;
|
||||
@ -41,11 +37,18 @@ import ch.psi.daq.query.model.Query;
|
||||
import ch.psi.daq.query.model.QueryField;
|
||||
import ch.psi.daq.query.model.impl.AbstractQuery;
|
||||
import ch.psi.daq.queryrest.controller.validator.QueryValidator;
|
||||
import ch.psi.daq.queryrest.filter.CorsFilter;
|
||||
import ch.psi.daq.queryrest.model.PropertyFilterMixin;
|
||||
import ch.psi.daq.queryrest.response.JsonByteArraySerializer;
|
||||
import ch.psi.daq.queryrest.response.JsonStreamSerializer;
|
||||
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
|
||||
|
||||
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;
|
||||
|
||||
@Configuration
|
||||
@PropertySource(value = {"classpath:queryrest.properties"})
|
||||
@PropertySource(value = {"file:${user.home}/.config/daq/queryrest.properties"}, ignoreResourceNotFound = true)
|
||||
@ -179,6 +182,12 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
|
||||
return new QueryValidator();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnProperty("queryrest.cors.enable")
|
||||
public Filter corsFilter() {
|
||||
return new CorsFilter();
|
||||
}
|
||||
|
||||
// ==========================================================================================
|
||||
// TODO: This is simply for initial / rudimentary testing - remove once further evolved
|
||||
@Bean
|
||||
|
||||
@ -2,9 +2,11 @@ package ch.psi.daq.queryrest.controller;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Collection;
|
||||
import java.util.List;
|
||||
import java.util.Map.Entry;
|
||||
import java.util.Set;
|
||||
import java.util.function.Function;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
@ -27,9 +29,11 @@ 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.common.ordering.Ordering;
|
||||
import ch.psi.daq.domain.DataEvent;
|
||||
import ch.psi.daq.query.analyzer.QueryAnalyzer;
|
||||
import ch.psi.daq.query.model.Aggregation;
|
||||
import ch.psi.daq.query.model.AggregationType;
|
||||
import ch.psi.daq.query.model.DBMode;
|
||||
import ch.psi.daq.query.model.Query;
|
||||
import ch.psi.daq.query.model.QueryField;
|
||||
@ -39,6 +43,8 @@ import ch.psi.daq.queryrest.config.QueryRestConfig;
|
||||
import ch.psi.daq.queryrest.model.ChannelsRequest;
|
||||
import ch.psi.daq.queryrest.response.ResponseStreamWriter;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
|
||||
@RestController
|
||||
public class QueryRestController {
|
||||
|
||||
@ -114,6 +120,68 @@ public class QueryRestController {
|
||||
return getChannels(new ChannelsRequest(channelName));
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Returns the current list of {@link Ordering}s available.
|
||||
*
|
||||
* @return list of {@link Ordering}s as String array
|
||||
*/
|
||||
@RequestMapping(value = "ordering", method = {RequestMethod.GET},
|
||||
produces = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public @ResponseBody List<String> getOrderingValues() {
|
||||
List<Ordering> orderings = Lists.newArrayList(Ordering.values());
|
||||
return orderings.stream()
|
||||
.map((Ordering ord) -> {
|
||||
return ord.toString();
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current list of {@link QueryField}s available.
|
||||
*
|
||||
* @return list of {@link QueryField}s as String array
|
||||
*/
|
||||
@RequestMapping(value = "queryfields", method = {RequestMethod.GET},
|
||||
produces = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public @ResponseBody List<String> getQueryFieldValues() {
|
||||
List<QueryField> orderings = Lists.newArrayList(QueryField.values());
|
||||
return orderings.stream()
|
||||
.map((QueryField qf) -> {
|
||||
return qf.toString();
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current list of {@link Aggregation}s available.
|
||||
*
|
||||
* @return list of {@link Aggregation}s as String array
|
||||
*/
|
||||
@RequestMapping(value = "aggregations", method = {RequestMethod.GET},
|
||||
produces = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public @ResponseBody List<String> getAggregationsValues() {
|
||||
List<Aggregation> orderings = Lists.newArrayList(Aggregation.values());
|
||||
return orderings.stream()
|
||||
.map((Aggregation value) -> {
|
||||
return value.toString();
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
/**
|
||||
* Returns the current list of {@link AggregationType}s available.
|
||||
*
|
||||
* @return list of {@link AggregationType}s as String array
|
||||
*/
|
||||
@RequestMapping(value = "aggregationtypes", method = {RequestMethod.GET},
|
||||
produces = {MediaType.APPLICATION_JSON_VALUE})
|
||||
public @ResponseBody List<String> getAggregationTypesValues() {
|
||||
List<AggregationType> orderings = Lists.newArrayList(AggregationType.values());
|
||||
return orderings.stream()
|
||||
.map((AggregationType value) -> {
|
||||
return value.toString();
|
||||
}).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* Catch-all query method for getting data from the backend.
|
||||
* <p>
|
||||
@ -143,8 +211,8 @@ public class QueryRestController {
|
||||
QueryAnalyzer queryAnalizer = queryAnalizerFactory.apply(query);
|
||||
|
||||
// all the magic happens here
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents = getQueryProcessor(query.getDbMode())
|
||||
.process(queryAnalizer);
|
||||
Stream<Entry<String, Stream<? extends DataEvent>>> channelToDataEvents =
|
||||
getQueryProcessor(query.getDbMode()).process(queryAnalizer);
|
||||
|
||||
// do post-process
|
||||
Stream<Entry<String, ?>> channelToData = queryAnalizer.postProcess(channelToDataEvents);
|
||||
|
||||
80
src/main/java/ch/psi/daq/queryrest/filter/CorsFilter.java
Normal file
80
src/main/java/ch/psi/daq/queryrest/filter/CorsFilter.java
Normal file
@ -0,0 +1,80 @@
|
||||
package ch.psi.daq.queryrest.filter;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.Arrays;
|
||||
import java.util.HashSet;
|
||||
import java.util.Set;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import javax.servlet.FilterChain;
|
||||
import javax.servlet.ServletException;
|
||||
import javax.servlet.http.HttpServletRequest;
|
||||
import javax.servlet.http.HttpServletResponse;
|
||||
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.web.filter.OncePerRequestFilter;
|
||||
|
||||
public class CorsFilter extends OncePerRequestFilter {
|
||||
|
||||
private static final String ALLOW_ORIGIN_HEADER = "Access-Control-Allow-Origin";
|
||||
|
||||
@Value("${queryrest.cors.allowedorigins}")
|
||||
private String configuredOrigins;
|
||||
|
||||
@Value("${queryrest.cors.forceallheaders}")
|
||||
private boolean forceAllHeaders;
|
||||
|
||||
|
||||
|
||||
/**
|
||||
* @{inheritDoc
|
||||
*/
|
||||
@Override
|
||||
protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
|
||||
throws ServletException, IOException {
|
||||
|
||||
Set<String> allowedOrigins = new HashSet<String>(Arrays.asList(configuredOrigins.split(","))
|
||||
.stream()
|
||||
.map(s -> {
|
||||
return s.trim(); })
|
||||
.collect(Collectors.toList()));
|
||||
|
||||
String originHeader = request.getHeader("Origin");
|
||||
if (forceAllHeaders) {
|
||||
// include headers no matter what - good for development
|
||||
if (allowedOrigins.contains(originHeader)) {
|
||||
response.addHeader(ALLOW_ORIGIN_HEADER, originHeader);
|
||||
} else {
|
||||
response.addHeader(ALLOW_ORIGIN_HEADER, "*");
|
||||
}
|
||||
setDefaultCorsHeaders(response);
|
||||
|
||||
} else if (request.getHeader("Access-Control-Request-Method") != null && "OPTIONS".equals(request.getMethod())) {
|
||||
// this is for 'real' Cross-site browser requests
|
||||
if (allowedOrigins.contains(originHeader)) {
|
||||
response.addHeader(ALLOW_ORIGIN_HEADER, originHeader);
|
||||
}
|
||||
setDefaultCorsHeaders(response);
|
||||
}
|
||||
|
||||
filterChain.doFilter(request, response);
|
||||
}
|
||||
|
||||
|
||||
private void setDefaultCorsHeaders(HttpServletResponse response) {
|
||||
response.addHeader("Access-Control-Allow-Methods", "GET, POST, PUT, DELETE, OPTIONS");
|
||||
response.addHeader("Access-Control-Allow-Headers", "Origin, Authorization, Accept, Content-Type");
|
||||
response.addHeader("Access-Control-Max-Age", "1800");
|
||||
}
|
||||
|
||||
|
||||
public void setConfiguredOrigins(String configuredOrigins) {
|
||||
this.configuredOrigins = configuredOrigins;
|
||||
}
|
||||
|
||||
|
||||
public void setForceAllHeaders(boolean forceAllHeaders) {
|
||||
this.forceAllHeaders = forceAllHeaders;
|
||||
}
|
||||
|
||||
}
|
||||
@ -1,4 +1,14 @@
|
||||
# defines the fields that are included in the response
|
||||
# if no fields have been specified by the user
|
||||
queryrest.default.response.fields=channel,pulseId,globalMillis,globalNanos,iocMillis,iocNanos,shape,value
|
||||
queryrest.default.response.aggregations=min,max,sum
|
||||
|
||||
# aggregation which are included in the response by default if aggregation is enabled for a given query
|
||||
queryrest.default.response.aggregations=min,max,sum
|
||||
|
||||
# enables / disables the CORS servlet filter. Adds multiple CORS headers to the response
|
||||
queryrest.cors.enable=true
|
||||
# includes the CORS headers no matter what request or preflight was sent. If an Origin header is set, this header will be used.
|
||||
# If no Origin header is set, '*' will be used.
|
||||
queryrest.cors.forceallheaders=true
|
||||
# defines the allowed origins for CORS requests. Only relevant if queryrest.enableCORS==true (see above).
|
||||
queryrest.cors.allowedorigins=http://localhost:8080, *
|
||||
@ -27,9 +27,9 @@ import ch.psi.daq.test.cassandra.CassandraDaqUnitDependencyInjectionTestExecutio
|
||||
CassandraDaqUnitDependencyInjectionTestExecutionListener.class,
|
||||
DependencyInjectionTestExecutionListener.class})
|
||||
@SpringApplicationConfiguration(classes = {
|
||||
QueryRestApplication.class
|
||||
,QueryRestConfig.class
|
||||
,DaqWebMvcConfig.class
|
||||
QueryRestApplication.class,
|
||||
QueryRestConfig.class,
|
||||
DaqWebMvcConfig.class
|
||||
})
|
||||
@DirtiesContext(classMode = ClassMode.AFTER_EACH_TEST_METHOD)
|
||||
@WebAppConfiguration
|
||||
|
||||
@ -9,13 +9,13 @@ import org.springframework.context.annotation.PropertySources;
|
||||
import org.springframework.web.servlet.config.annotation.EnableWebMvc;
|
||||
import org.springframework.web.servlet.config.annotation.WebMvcConfigurationSupport;
|
||||
|
||||
import ch.psi.daq.domain.reader.DataReader;
|
||||
import ch.psi.daq.cassandra.reader.CassandraReader;
|
||||
import ch.psi.daq.query.processor.QueryProcessor;
|
||||
import ch.psi.daq.query.processor.QueryProcessorLocal;
|
||||
import ch.psi.daq.test.cassandra.admin.CassandraTestAdmin;
|
||||
import ch.psi.daq.test.cassandra.admin.CassandraTestAdminImpl;
|
||||
import ch.psi.daq.test.query.config.LocalQueryTestConfig;
|
||||
import ch.psi.daq.test.queryrest.query.DummyDataReader;
|
||||
import ch.psi.daq.test.queryrest.query.DummyCassandraReader;
|
||||
|
||||
@Configuration
|
||||
@ComponentScan
|
||||
@ -32,13 +32,12 @@ public class DaqWebMvcConfig extends WebMvcConfigurationSupport {
|
||||
}
|
||||
|
||||
@Bean
|
||||
public QueryProcessor queryProcessor() {
|
||||
return new QueryProcessorLocal(new DummyDataReader());
|
||||
public QueryProcessor cassandraQueryProcessorLocal() {
|
||||
return new QueryProcessorLocal(cassandraReader());
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DataReader dataReader() {
|
||||
return new DummyDataReader();
|
||||
@Bean CassandraReader cassandraReader() {
|
||||
return new DummyCassandraReader();
|
||||
}
|
||||
|
||||
@Bean
|
||||
|
||||
@ -1,204 +0,0 @@
|
||||
package ch.psi.daq.test.queryrest.controller;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
|
||||
import org.springframework.test.web.servlet.result.MockMvcResultHandlers;
|
||||
import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
|
||||
|
||||
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
|
||||
import ch.psi.daq.common.ordering.Ordering;
|
||||
import ch.psi.daq.query.model.AggregationType;
|
||||
import ch.psi.daq.query.model.impl.PulseRangeQuery;
|
||||
import ch.psi.daq.query.model.impl.TimeRangeQuery;
|
||||
import ch.psi.daq.query.model.impl.TimeRangeQueryDate;
|
||||
import ch.psi.daq.queryrest.controller.QueryRestController;
|
||||
import ch.psi.daq.test.cassandra.admin.CassandraTestAdmin;
|
||||
import ch.psi.daq.test.queryrest.AbstractDaqRestTest;
|
||||
|
||||
/**
|
||||
* Tests the {@link DaqController} implementation.
|
||||
*/
|
||||
public class DaqRestControllerTest extends AbstractDaqRestTest {
|
||||
|
||||
@Resource
|
||||
private CassandraTestAdmin cassandraTestAdmin;
|
||||
|
||||
@Resource
|
||||
private CassandraDataGen dataGen;
|
||||
|
||||
private static final boolean initialized = false;
|
||||
|
||||
private static final int DATA_KEYSPACE = 1;
|
||||
public static final String[] TEST_CHANNEL_NAMES = new String[]{"testChannel1", "testChannel2"};
|
||||
|
||||
@Before
|
||||
public void setUp() throws Exception {
|
||||
if (!initialized) {
|
||||
cassandraTestAdmin.truncateAll();
|
||||
|
||||
dataGen.writeData(DATA_KEYSPACE, 100, 2, TEST_CHANNEL_NAMES);
|
||||
}
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {}
|
||||
|
||||
@Test
|
||||
public void testChannelNameQuery() throws Exception {
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.get(QueryRestController.CHANNELS)
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[2]").doesNotExist());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testPulseRangeQuery() throws Exception {
|
||||
PulseRangeQuery request = new PulseRangeQuery(
|
||||
100,
|
||||
101,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
System.out.println(content);
|
||||
|
||||
content = "{\"channels\":[\"testChannel1\",\"testChannel2\"],\"startPulseId\":100,\"endPulseId\":101}";
|
||||
|
||||
this.mockMvc
|
||||
.perform(MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTimeRangeQuery() throws Exception {
|
||||
TimeRangeQuery request = new TimeRangeQuery(
|
||||
100,
|
||||
101,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
|
||||
this.mockMvc.perform(MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDateRangeQuery() throws Exception {
|
||||
String startDate = TimeRangeQueryDate.format(100);
|
||||
String endDate = TimeRangeQueryDate.format(101);
|
||||
TimeRangeQueryDate request = new TimeRangeQueryDate(
|
||||
startDate,
|
||||
endDate,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
System.out.println(content);
|
||||
|
||||
this.mockMvc
|
||||
.perform(
|
||||
MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content)
|
||||
)
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExtremaAggregation() throws Exception {
|
||||
PulseRangeQuery request = new PulseRangeQuery(
|
||||
100,
|
||||
101,
|
||||
false,
|
||||
Ordering.asc,
|
||||
AggregationType.extrema,
|
||||
TEST_CHANNEL_NAMES[0]);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
|
||||
this.mockMvc
|
||||
.perform(MockMvcRequestBuilders.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.value").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.value").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.event").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.event.channel").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.event.channel").value(TEST_CHANNEL_NAMES[0]))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.event.pulseId").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.min.event.pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.maxima").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.value").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.value").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.event").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.event.channel").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.event.channel").value(TEST_CHANNEL_NAMES[0]))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.event.pulseId").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$.minima.max.event.pulseId").value(101));
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,299 @@
|
||||
package ch.psi.daq.test.queryrest.controller;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Test;
|
||||
import org.springframework.http.MediaType;
|
||||
import org.springframework.test.web.servlet.request.MockMvcRequestBuilders;
|
||||
import org.springframework.test.web.servlet.result.MockMvcResultHandlers;
|
||||
import org.springframework.test.web.servlet.result.MockMvcResultMatchers;
|
||||
import org.springframework.test.web.servlet.setup.MockMvcBuilders;
|
||||
|
||||
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
|
||||
import ch.psi.daq.common.ordering.Ordering;
|
||||
import ch.psi.daq.query.model.AggregationType;
|
||||
import ch.psi.daq.query.model.impl.PulseRangeQuery;
|
||||
import ch.psi.daq.query.model.impl.TimeRangeQuery;
|
||||
import ch.psi.daq.query.model.impl.TimeRangeQueryDate;
|
||||
import ch.psi.daq.queryrest.controller.QueryRestController;
|
||||
import ch.psi.daq.queryrest.filter.CorsFilter;
|
||||
import ch.psi.daq.test.cassandra.admin.CassandraTestAdmin;
|
||||
import ch.psi.daq.test.queryrest.AbstractDaqRestTest;
|
||||
|
||||
/**
|
||||
* Tests the {@link DaqController} implementation.
|
||||
*/
|
||||
public class QueryRestControllerTest extends AbstractDaqRestTest {
|
||||
|
||||
@Resource
|
||||
private CassandraTestAdmin cassandraTestAdmin;
|
||||
|
||||
@Resource
|
||||
private CassandraDataGen dataGen;
|
||||
|
||||
@Resource
|
||||
private CorsFilter corsFilter;
|
||||
|
||||
public static final String[] TEST_CHANNEL_NAMES = new String[] {"testChannel1", "testChannel2"};
|
||||
|
||||
@After
|
||||
public void tearDown() throws Exception {}
|
||||
|
||||
@Test
|
||||
public void testChannelNameQuery() throws Exception {
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.get(QueryRestController.CHANNELS)
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").value("BooleanScalar"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").value("BooleanWaveform"));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testSpecificChannelSearch() throws Exception {
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.get(QueryRestController.CHANNELS + "/integer")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").value("IntegerScalar"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").value("IntegerWaveform"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[2]").value("UIntegerScalar"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[3]").value("UIntegerWaveform"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[4]").doesNotExist());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCorsFilterNoHeaders() throws Exception {
|
||||
corsFilter.setForceAllHeaders(false);
|
||||
this.mockMvc = MockMvcBuilders.webAppContextSetup(webApplicationContext).addFilters(corsFilter).build();
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
// we didn't set the 'Origin' header so no access-control
|
||||
.andExpect(MockMvcResultMatchers.header().doesNotExist("Access-Control-Allow-Origin"));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCorsFilterIncludesHeaders() throws Exception {
|
||||
// all headers are set
|
||||
this.mockMvc = MockMvcBuilders.webAppContextSetup(webApplicationContext).addFilters(corsFilter).build();
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.header("Origin", "*")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
// we didn't set the 'Origin' header so no access-control
|
||||
.andExpect(MockMvcResultMatchers.header().string("Access-Control-Allow-Origin", "*"));
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.header("Origin", "http://localhost:8080")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
// we didn't set the 'Origin' header so no access-control
|
||||
.andExpect(MockMvcResultMatchers.header().string("Access-Control-Allow-Origin", "http://localhost:8080"));
|
||||
|
||||
this.mockMvc.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.header("Origin", "someBogusDomain.com")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.header().string("Access-Control-Allow-Origin", "*"));
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testCorsFilterMismatchSpecificOrigin() throws Exception {
|
||||
corsFilter.setForceAllHeaders(false);
|
||||
this.mockMvc = MockMvcBuilders.webAppContextSetup(webApplicationContext).addFilters(corsFilter).build();
|
||||
|
||||
this.mockMvc
|
||||
.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.header("Origin", "*")
|
||||
.header("Access-Control-Request-Method", "GET")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.header().string("Access-Control-Allow-Origin", "*"))
|
||||
.andExpect(
|
||||
MockMvcResultMatchers.header().string("Access-Control-Allow-Headers",
|
||||
"Origin, Authorization, Accept, Content-Type"));
|
||||
|
||||
this.mockMvc
|
||||
.perform(
|
||||
MockMvcRequestBuilders
|
||||
.options(QueryRestController.CHANNELS)
|
||||
.header("Origin", "someBogusDomain.com")
|
||||
.header("Access-Control-Request-Method", "GET")
|
||||
.contentType(MediaType.APPLICATION_JSON))
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.header().doesNotExist("Access-Control-Allow-Origin"))
|
||||
.andExpect(
|
||||
MockMvcResultMatchers.header().string("Access-Control-Allow-Headers",
|
||||
"Origin, Authorization, Accept, Content-Type"));
|
||||
}
|
||||
|
||||
|
||||
@Test
|
||||
public void testPulseRangeQuery() throws Exception {
|
||||
PulseRangeQuery request = new PulseRangeQuery(
|
||||
100,
|
||||
101,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
System.out.println(content);
|
||||
|
||||
content = "{\"channels\":[\"testChannel1\",\"testChannel2\"],\"startPulseId\":100,\"endPulseId\":101}";
|
||||
|
||||
this.mockMvc
|
||||
.perform(MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testTimeRangeQuery() throws Exception {
|
||||
TimeRangeQuery request = new TimeRangeQuery(
|
||||
100,
|
||||
101,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
|
||||
this.mockMvc.perform(MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testDateRangeQuery() throws Exception {
|
||||
String startDate = TimeRangeQueryDate.format(100);
|
||||
String endDate = TimeRangeQueryDate.format(101);
|
||||
TimeRangeQueryDate request = new TimeRangeQueryDate(
|
||||
startDate,
|
||||
endDate,
|
||||
TEST_CHANNEL_NAMES);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
System.out.println(content);
|
||||
|
||||
this.mockMvc
|
||||
.perform(
|
||||
MockMvcRequestBuilders
|
||||
.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content)
|
||||
)
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value("testChannel1"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data[1].pulseId").value(101))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1]").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].channel").value("testChannel2"))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data").isArray())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[0].pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[1].data[1].pulseId").value(101));
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testExtremaAggregation() throws Exception {
|
||||
PulseRangeQuery request = new PulseRangeQuery(
|
||||
100,
|
||||
101,
|
||||
false,
|
||||
Ordering.asc,
|
||||
AggregationType.extrema,
|
||||
TEST_CHANNEL_NAMES[0]);
|
||||
|
||||
String content = mapper.writeValueAsString(request);
|
||||
|
||||
this.mockMvc
|
||||
.perform(MockMvcRequestBuilders.post(QueryRestController.QUERY)
|
||||
.contentType(MediaType.APPLICATION_JSON)
|
||||
.content(content))
|
||||
|
||||
.andDo(MockMvcResultHandlers.print())
|
||||
.andExpect(MockMvcResultMatchers.status().isOk())
|
||||
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].channel").value(TEST_CHANNEL_NAMES[0]))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min.value").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min.value").value(Double.valueOf(100)))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min.event").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min.event.pulseId").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.minima.min.event.pulseId").value(100))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max.value").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max.value").value(Double.valueOf(101)))
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max.event").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max.event.pulseId").exists())
|
||||
.andExpect(MockMvcResultMatchers.jsonPath("$[0].data.maxima.max.event.pulseId").value(101));
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@ -0,0 +1,252 @@
|
||||
/**
|
||||
*
|
||||
*/
|
||||
package ch.psi.daq.test.queryrest.query;
|
||||
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CompletableFuture;
|
||||
import java.util.regex.Pattern;
|
||||
import java.util.stream.Collectors;
|
||||
import java.util.stream.Stream;
|
||||
|
||||
import org.apache.commons.lang.NotImplementedException;
|
||||
|
||||
import ch.psi.daq.cassandra.reader.CassandraReader;
|
||||
import ch.psi.daq.cassandra.reader.query.PulseIdRangeQuery;
|
||||
import ch.psi.daq.cassandra.reader.query.TimeRangeQuery;
|
||||
import ch.psi.daq.cassandra.util.test.CassandraDataGen;
|
||||
import ch.psi.daq.common.ordering.Ordering;
|
||||
import ch.psi.daq.domain.DataEvent;
|
||||
import ch.psi.daq.domain.cassandra.ChannelEvent;
|
||||
import ch.psi.daq.domain.cassandra.MetaPulseId;
|
||||
import ch.psi.daq.domain.cassandra.MetaTime;
|
||||
import ch.psi.daq.domain.cassandra.querying.ChannelEventQuery;
|
||||
import ch.psi.daq.domain.cassandra.querying.ChannelEventQueryInfo;
|
||||
import ch.psi.daq.domain.cassandra.querying.EventQuery;
|
||||
|
||||
import com.google.common.collect.Lists;
|
||||
|
||||
/**
|
||||
* @author zellweger_c
|
||||
*
|
||||
*/
|
||||
public class DummyCassandraReader implements CassandraReader {
|
||||
|
||||
private static final int KEYSPACE = 1;
|
||||
private CassandraDataGen dataGen;
|
||||
private String[] channels;
|
||||
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
public DummyCassandraReader() {
|
||||
this.dataGen = new CassandraDataGen();
|
||||
this.channels = new String[]{
|
||||
"testChannel1",
|
||||
"testChannel2",
|
||||
"BooleanScalar",
|
||||
"BooleanWaveform",
|
||||
"ByteScalar",
|
||||
"ByteWaveform",
|
||||
"DoubleScalar",
|
||||
"DoubleWaveform",
|
||||
"FloatScalar",
|
||||
"FloatWaveform",
|
||||
"IntegerScalar",
|
||||
"IntegerWaveform",
|
||||
"LongScalar",
|
||||
"LongWaveform",
|
||||
"ShortScalar",
|
||||
"ShortWaveform",
|
||||
"StringScalar",
|
||||
"UByteScalar",
|
||||
"UByteWaveform",
|
||||
"UIntegerScalar",
|
||||
"UIntegerWaveform",
|
||||
"ULongScalar",
|
||||
"ULongWaveform",
|
||||
"UShortScalar",
|
||||
"UShortWaveform"};
|
||||
}
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<String> getChannelStream(String regex) {
|
||||
Stream<String> channelStream = Stream.of(channels);
|
||||
if (regex != null) {
|
||||
Pattern pattern = Pattern.compile(regex.toLowerCase());
|
||||
channelStream = Stream.of(channels).filter(channel -> pattern.matcher(channel.toLowerCase()).find());
|
||||
}
|
||||
return channelStream;
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<? extends DataEvent> getEventStream(String channel, long startPulseId, long endPulseId,
|
||||
Ordering ordering, String... columns) {
|
||||
return getDummyEventStream(channel, startPulseId, endPulseId);
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<? extends DataEvent> getEventStream(String channel, long startMillis, long startNanos, long endMillis,
|
||||
long endNanos, Ordering ordering, String... columns) {
|
||||
return getDummyEventStream(channel, startMillis, endMillis);
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<ChannelEvent> getEventStream(EventQuery eventQuery, Stream<? extends ChannelEventQuery> queryProviders) {
|
||||
List<ChannelEvent> result = Lists.newArrayList();
|
||||
queryProviders.forEach(ceq -> {
|
||||
if (ceq instanceof ChannelEventQueryInfo) {
|
||||
result.add(getEvent((ChannelEventQueryInfo) ceq));
|
||||
} else {
|
||||
throw new NotImplementedException("This is not yet implemented!");
|
||||
}
|
||||
});
|
||||
return result.stream();
|
||||
}
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<ChannelEvent> getEventStream(PulseIdRangeQuery query) {
|
||||
Stream<ChannelEvent> dummyEventStream = getDummyEventStream(query.getChannel(), query.getStartPulseId(), query.getEndPulseId())
|
||||
.map(ce -> { return (ChannelEvent) ce; });
|
||||
return dummyEventStream;
|
||||
|
||||
}
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<ChannelEvent> getEventStream(TimeRangeQuery query) {
|
||||
Stream<ChannelEvent> dummyEventStream = getDummyEventStream(query.getChannel(), query.getStartMillis(), query.getEndMillis())
|
||||
.map(ce -> { return (ChannelEvent) ce; });
|
||||
return dummyEventStream;
|
||||
}
|
||||
|
||||
|
||||
private Stream<? extends DataEvent> getDummyEventStream(String channel, long startIndex, long endIndex) {
|
||||
|
||||
return dataGen.generateData(startIndex, (endIndex-startIndex + 1),
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> Long.valueOf(i),
|
||||
channel).stream();
|
||||
}
|
||||
|
||||
private List<? extends DataEvent> getDummyEvents(String channel, long startIndex, long endIndex) {
|
||||
|
||||
return dataGen.generateData(startIndex, (endIndex-startIndex + 1),
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> Long.valueOf(i),
|
||||
channel);
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public List<String> getChannels() {
|
||||
return Lists.newArrayList(channels);
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public List<String> getChannels(String regex) {
|
||||
return Lists.newArrayList(channels).stream().filter(s -> { return s.contains(regex); }).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public ChannelEvent getEvent(ChannelEventQueryInfo queryInfo, String... columns) {
|
||||
if (queryInfo.getPulseId() > 0) {
|
||||
return (ChannelEvent) getDummyEvents(queryInfo.getChannel(), queryInfo.getPulseId(), queryInfo.getPulseId()).get(0);
|
||||
}
|
||||
return (ChannelEvent) getDummyEvents(queryInfo.getChannel(), queryInfo.getGlobalMillis(), queryInfo.getGlobalMillis()).get(0);
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public CompletableFuture<ChannelEvent> getEventAsync(ChannelEventQueryInfo queryInfo, String... columns) {
|
||||
throw new NotImplementedException();
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<? extends ChannelEventQuery> getChannelEventQueryStream(PulseIdRangeQuery query) {
|
||||
|
||||
|
||||
return dataGen.generateMetaPulseId(
|
||||
KEYSPACE,
|
||||
query.getStartPulseId(),
|
||||
(query.getEndPulseId() - query.getStartPulseId() + 1),
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
query.getChannel()).stream();
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<? extends ChannelEventQuery> getChannelEventQueryStream(TimeRangeQuery query) {
|
||||
|
||||
|
||||
return dataGen.generateMetaTime(
|
||||
KEYSPACE,
|
||||
query.getStartMillis(),
|
||||
(query.getEndMillis() - query.getStartMillis() + 1),
|
||||
i -> i,
|
||||
i -> i,
|
||||
i -> i,
|
||||
query.getChannel()).stream();
|
||||
}
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<MetaPulseId> getMetaPulseIdStream(PulseIdRangeQuery query) {
|
||||
|
||||
return getChannelEventQueryStream(query).map(r -> { return (MetaPulseId) r; });
|
||||
|
||||
}
|
||||
|
||||
|
||||
/**
|
||||
* @{inheritDoc}
|
||||
*/
|
||||
@Override
|
||||
public Stream<MetaTime> getMetaTimeStream(TimeRangeQuery query) {
|
||||
|
||||
return getChannelEventQueryStream(query).map(r -> { return (MetaTime) r; });
|
||||
}
|
||||
|
||||
}
|
||||
Reference in New Issue
Block a user