ATEST-299

src/main/java/ch/psi/daq/queryrest/config/QueryRestConfig.java

@@ -42,9 +42,10 @@ import ch.psi.daq.query.model.QueryField;
 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 ch.psi.daq.queryrest.response.csv.CSVResponseStreamWriter;
+import ch.psi.daq.queryrest.response.json.JSONResponseStreamWriter;
+import ch.psi.daq.queryrest.response.json.JsonByteArraySerializer;
+import ch.psi.daq.queryrest.response.json.JsonStreamSerializer;
 
 @Configuration
 @PropertySource(value = {"classpath:queryrest.properties"})
@@ -122,8 +123,13 @@ public class QueryRestConfig extends WebMvcConfigurerAdapter {
    }
 
    @Bean
-   public ResponseStreamWriter responseStreamWriter() {
-      return new ResponseStreamWriter();
+   public JSONResponseStreamWriter jsonResponseStreamWriter() {
+      return new JSONResponseStreamWriter();
+   }
+   
+   @Bean
+   public CSVResponseStreamWriter csvResponseStreamWriter() {
+      return new CSVResponseStreamWriter();
    }
 
    @Bean(name = BEAN_NAME_DEFAULT_RESPONSE_FIELDS)

src/main/java/ch/psi/daq/queryrest/controller/QueryRestController.java

@@ -41,7 +41,8 @@ import ch.psi.daq.query.model.QueryField;
 import ch.psi.daq.query.model.impl.DAQQuery;
 import ch.psi.daq.query.processor.QueryProcessor;
 import ch.psi.daq.query.request.ChannelsRequest;
-import ch.psi.daq.queryrest.response.ResponseStreamWriter;
+import ch.psi.daq.queryrest.response.csv.CSVResponseStreamWriter;
+import ch.psi.daq.queryrest.response.json.JSONResponseStreamWriter;
 
 import com.google.common.collect.Lists;
 
@@ -58,7 +59,10 @@ public class QueryRestController {
    private Validator requestProviderValidator = new RequestProviderValidator();
 
    @Resource
-   private ResponseStreamWriter responseStreamWriter;
+   private JSONResponseStreamWriter jsonResponseStreamWriter;
+
+   @Resource
+   private CSVResponseStreamWriter csvResponseStreamWriter;
 
    @Resource
    private ApplicationContext appContext;
@@ -190,9 +194,8 @@ public class QueryRestController {
             }).collect(Collectors.toList());
    }
 
-
    /**
-    * Catch-all query method for getting data from the backend.
+    * Catch-all query method for getting data from the backend for JSON requests.
     * <p>
     * The {@link DAQQuery} 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
@@ -203,14 +206,56 @@ public class QueryRestController {
     * @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 DAQQuery query, HttpServletResponse res) throws IOException {
+   @RequestMapping(
+         value = QUERY,
+         method = RequestMethod.POST,
+         consumes = {MediaType.APPLICATION_JSON_VALUE},
+         produces = {MediaType.APPLICATION_JSON_VALUE})
+   public void executeQueryJson(@RequestBody @Valid DAQQuery query, HttpServletResponse res) throws Exception {
       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) {
+         jsonResponseStreamWriter.respond(executeQuery(query), query, res);
+      } catch (Exception t) {
+         LOGGER.error("Failed to execute query '{}'.", query, t);
+         throw t;
+      }
+   }
+
+   /**
+    * Catch-all query method for getting data from the backend for JSON requests.
+    * <p>
+    * The {@link DAQQuery} 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 DAQQuery}
+    * @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},
+         produces = {CSVResponseStreamWriter.APPLICATION_CSV_VALUE})
+   public void executeQueryCsv(@RequestBody @Valid DAQQuery query, HttpServletResponse res) throws Exception {
+      if (!(query.getAggregationType() == null || AggregationType.value.equals(query.getAggregationType()))) {
+         // We allow only no aggregation or value aggregation as
+         // extrema: nested structure and not clear how to map it to one line
+         // index: value is an array of Statistics whose size is not clear at initialization time
+         String message = "CSV export does not support '" + query.getAggregationType() + "'";
+         LOGGER.warn(message);
+         throw new IllegalArgumentException(message);
+      }
+
+      try {
+         LOGGER.debug("Executing query '{}'", query.toString());
+
+         // write the response back to the client using java 8 streams
+         csvResponseStreamWriter.respond(executeQuery(query), query, res);
+      } catch (Exception t) {
          LOGGER.error("Failed to execute query '{}'.", query, t);
          throw t;
       }

src/main/java/ch/psi/daq/queryrest/controller/validator/QueryValidator.java

@@ -14,7 +14,6 @@ import ch.psi.daq.query.model.DBMode;
 import ch.psi.daq.query.model.QueryField;
 import ch.psi.daq.query.model.impl.DAQQuery;
 import ch.psi.daq.cassandra.request.Request;
-import ch.psi.daq.cassandra.request.range.RequestRangeTime;
 import ch.psi.daq.queryrest.config.QueryRestConfig;
 
 public class QueryValidator implements Validator {

src/main/java/ch/psi/daq/queryrest/response/ResponseStreamWriter.java

@@ -1,46 +1,14 @@
 package ch.psi.daq.queryrest.response;
 
 import java.io.IOException;
-import java.util.LinkedHashSet;
-import java.util.List;
-import java.util.Set;
 import java.util.Map.Entry;
 import java.util.stream.Stream;
 
-import javax.annotation.Resource;
 import javax.servlet.ServletResponse;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.http.MediaType;
-
-import com.fasterxml.jackson.core.JsonEncoding;
-import com.fasterxml.jackson.core.JsonFactory;
-import com.fasterxml.jackson.core.JsonGenerator;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import com.fasterxml.jackson.databind.ObjectWriter;
-import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
-import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
-
-import ch.psi.daq.query.model.Aggregation;
-import ch.psi.daq.query.model.QueryField;
 import ch.psi.daq.query.model.impl.DAQQuery;
 
-/**
- * Takes a Java 8 stream and writes it to the output stream provided by the {@link ServletResponse}
- * of the current request.
- */
-public class ResponseStreamWriter {
-
-   private static final String DATA_RESP_FIELD = "data";
-
-   private static final Logger logger = LoggerFactory.getLogger(ResponseStreamWriter.class);
-
-   @Resource
-   private JsonFactory jsonFactory;
-
-   @Resource
-   private ObjectMapper mapper;
+public interface ResponseStreamWriter {
 
    /**
     * Responding with the the contents of the stream by writing into the output stream of the
@@ -51,82 +19,5 @@ public class ResponseStreamWriter {
     * @param response {@link ServletResponse} instance given by the current HTTP request
     * @throws IOException thrown if writing to the output stream fails
     */
-   public void respond(Stream<Entry<String, ?>> stream, DAQQuery query,
-         ServletResponse response) throws IOException {
-
-      Set<QueryField> queryFields = query.getFields();
-      List<Aggregation> aggregations = query.getAggregations();
-
-      Set<String> includedFields =
-            new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
-
-      for (QueryField field : queryFields) {
-         includedFields.add(field.name());
-      }
-      if (aggregations != null) {
-         for (Aggregation aggregation : query.getAggregations()) {
-            includedFields.add(aggregation.name());
-         }
-      }
-      // do not write channel since it is already provided as key in mapping
-      includedFields.remove(QueryField.channel.name());
-
-      ObjectWriter writer = configureWriter(includedFields);
-      respondInternal(stream, response, writer);
-   }
-
-   /**
-    * Configures the writer dynamically by including the fields which should be included in the
-    * response.
-    * 
-    * @param includedFields set of strings which correspond to the getter method names of the
-    *        classes registered as a mixed-in
-    * @return the configured writer that includes the specified fields
-    */
-   private ObjectWriter configureWriter(Set<String> includedFields) {
-      SimpleFilterProvider propertyFilter = new SimpleFilterProvider();
-      propertyFilter.addFilter("namedPropertyFilter", SimpleBeanPropertyFilter.filterOutAllExcept(includedFields));
-      // only write the properties not excluded in the filter
-      ObjectWriter writer = mapper.writer(propertyFilter);
-      return writer;
-   }
-   
-   /**
-    * Writes the outer Java stream into the output stream.
-    * 
-    * @param stream Mapping from channel name to data
-    * @param response {@link ServletResponse} instance given by the current HTTP request
-    * @param writer configured writer that includes the fields the end user wants to see
-    * @throws IOException thrown if writing to the output stream fails
-    */
-   private void respondInternal(Stream<Entry<String, ?>> stream, ServletResponse response,
-         ObjectWriter writer)
-         throws IOException {
-
-      response.setCharacterEncoding(JsonEncoding.UTF8.getJavaName());
-      response.setContentType(MediaType.APPLICATION_JSON_VALUE);
-      JsonGenerator generator = jsonFactory.createGenerator(response.getOutputStream(), JsonEncoding.UTF8);
-      generator.writeStartArray();
-      stream
-            /* ensure elements are sequentially written */
-            .sequential()
-            .forEach(
-                  entry -> {
-                     try {
-                        generator.writeStartObject();
-                        generator.writeStringField(QueryField.channel.name(), entry.getKey());
-
-                        generator.writeFieldName(DATA_RESP_FIELD);
-                        writer.writeValue(generator, entry.getValue());
-                        
-                        generator.writeEndObject();
-                     } catch (Exception e) {
-                        logger.error("Could not write channel name of channel '{}'", entry.getKey(), e);
-                     }
-                  });
-      generator.writeEndArray();
-      generator.flush();
-      generator.close();
-   }
-
+   public void respond(Stream<Entry<String, ?>> stream, DAQQuery query, ServletResponse response) throws Exception;
 }

src/main/java/ch/psi/daq/queryrest/response/csv/ArrayProcessor.java

@@ -0,0 +1,58 @@
+package ch.psi.daq.queryrest.response.csv;
+
+import java.lang.reflect.Array;
+
+import org.supercsv.cellprocessor.CellProcessorAdaptor;
+import org.supercsv.cellprocessor.ift.CellProcessor;
+import org.supercsv.util.CsvContext;
+
+public class ArrayProcessor extends CellProcessorAdaptor {
+   public static final char DEFAULT_SEPARATOR = ',';
+   public static final String OPEN_BRACKET = "[";
+   public static final String CLOSE_BRACKET = "]";
+
+   private char separator = DEFAULT_SEPARATOR;
+
+   public ArrayProcessor() {
+      super();
+   }
+
+   public ArrayProcessor(char separator) {
+      super();
+      this.separator = separator;
+   }
+
+   public ArrayProcessor(CellProcessor next) {
+      super(next);
+   }
+
+   public ArrayProcessor(CellProcessor next, char separator) {
+      super(next);
+      this.separator = separator;
+   }
+
+   @SuppressWarnings("unchecked")
+   @Override
+   public Object execute(Object value, CsvContext context) {
+      if (value.getClass().isArray()) {
+         StringBuilder buf = new StringBuilder();
+         
+         int length = Array.getLength(value);
+         buf.append(OPEN_BRACKET);
+         for (int i = 0; i < length;) {
+            Object val = next.execute(Array.get(value, i), context);
+            buf.append(val);
+
+            ++i;
+            if (i < length) {
+               buf.append(separator);
+            }
+         }
+         buf.append(CLOSE_BRACKET);
+
+         return buf.toString();
+      } else {
+         return next.execute(value, context);
+      }
+   }
+}

src/main/java/ch/psi/daq/queryrest/response/csv/CSVResponseStreamWriter.java

@@ -0,0 +1,140 @@
+package ch.psi.daq.queryrest.response.csv;
+
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map.Entry;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import javax.annotation.Resource;
+import javax.servlet.ServletResponse;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.supercsv.cellprocessor.constraint.NotNull;
+import org.supercsv.cellprocessor.ift.CellProcessor;
+import org.supercsv.io.dozer.CsvDozerBeanWriter;
+import org.supercsv.io.dozer.ICsvDozerBeanWriter;
+import org.supercsv.prefs.CsvPreference;
+
+import com.fasterxml.jackson.core.JsonEncoding;
+
+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.Query;
+import ch.psi.daq.query.model.QueryField;
+import ch.psi.daq.query.model.impl.DAQQuery;
+import ch.psi.daq.queryrest.response.ResponseStreamWriter;
+
+/**
+ * Takes a Java 8 stream and writes it to the output stream provided by the {@link ServletResponse}
+ * of the current request.
+ */
+public class CSVResponseStreamWriter implements ResponseStreamWriter {
+   public static final String APPLICATION_CSV_VALUE = "text/csv";
+
+   private static final char DELIMITER_CVS = ';';
+   private static final char DELIMITER_ARRAY = ',';
+
+   private static final Logger LOGGER = LoggerFactory.getLogger(CSVResponseStreamWriter.class);
+   
+   @Resource
+   private Function<Query, QueryAnalyzer> queryAnalizerFactory;
+
+   @Override
+   public void respond(Stream<Entry<String, ?>> stream, DAQQuery query, ServletResponse response) throws Exception {
+      response.setCharacterEncoding(JsonEncoding.UTF8.getJavaName());
+      response.setContentType(APPLICATION_CSV_VALUE);
+
+      respondInternal(stream, query, response);
+   }
+
+   private void respondInternal(Stream<Entry<String, ?>> stream, DAQQuery query, ServletResponse response)
+         throws Exception {
+      Set<QueryField> queryFields = query.getFields();
+      List<Aggregation> aggregations = query.getAggregations();
+
+      Set<String> fieldMapping =
+            new LinkedHashSet<>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
+      List<String> header =
+            new ArrayList<>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
+      List<CellProcessor> processorSet =
+            new ArrayList<>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
+      boolean isNewField;
+      QueryAnalyzer queryAnalyzer = queryAnalizerFactory.apply(query);
+
+      for (QueryField field : queryFields) {
+         if(!(QueryField.value.equals(field) && queryAnalyzer.isAggregationEnabled())){
+            isNewField = fieldMapping.add(field.name());
+
+            if (isNewField) {
+               header.add(field.name());
+               processorSet.add(new ArrayProcessor(new NotNull(), DELIMITER_ARRAY));
+            }
+         }
+      }
+      if (aggregations != null && queryAnalyzer.isAggregationEnabled()) {
+         for (Aggregation aggregation : query.getAggregations()) {
+            isNewField = fieldMapping.add("value." + aggregation.name());
+
+            if (isNewField) {
+               header.add(aggregation.name());
+               processorSet.add(new ArrayProcessor(new NotNull(), DELIMITER_ARRAY));
+            }
+         }
+      }
+
+      CellProcessor[] processors = processorSet.toArray(new CellProcessor[processorSet.size()]);
+      CsvPreference preference = new CsvPreference.Builder(
+            CsvPreference.EXCEL_NORTH_EUROPE_PREFERENCE.getQuoteChar(),
+            DELIMITER_CVS,
+            CsvPreference.EXCEL_NORTH_EUROPE_PREFERENCE.getEndOfLineSymbols()).build();
+      ICsvDozerBeanWriter beanWriter = new CsvDozerBeanWriter(response.getWriter(), preference);
+
+      AtomicReference<Exception> exception = new AtomicReference<>();
+      try {
+         // configure the mapping from the fields to the CSV columns
+         beanWriter.configureBeanMapping(DataEvent.class, fieldMapping.toArray(new String[fieldMapping.size()]));
+
+         beanWriter.writeHeader(header.toArray(new String[header.size()]));
+
+         stream
+               /* ensure elements are sequentially written */
+               .sequential()
+               .forEach(
+                     entry -> {
+                        if (entry.getValue() instanceof Stream) {
+
+                           @SuppressWarnings("unchecked")
+                           Stream<DataEvent> eventStream = (Stream<DataEvent>) entry.getValue();
+                           eventStream
+                                 .forEach(
+                                 event -> {
+                                    try {
+                                       beanWriter.write(event, processors);
+                                    } catch (Exception e) {
+                                       LOGGER.error("Could not write elements for '{}-{}'", event.getChannel(),
+                                             event.getPulseId(), e);
+                                       exception.compareAndSet(null, e);
+                                    }
+                                 });
+                        } else {
+                           String message = "Type '" + entry.getValue().getClass() + "' not supported.";
+                           LOGGER.error(message);
+                           exception.compareAndSet(null, new RuntimeException(message));
+                        }
+                     });
+      } finally {
+         beanWriter.close();
+      }
+
+      if (exception.get() != null) {
+         throw exception.get();
+      }
+   }
+
+}

src/main/java/ch/psi/daq/queryrest/response/json/JSONResponseStreamWriter.java

@@ -0,0 +1,140 @@
+package ch.psi.daq.queryrest.response.json;
+
+import java.io.IOException;
+import java.util.Map.Entry;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Stream;
+
+import javax.annotation.Resource;
+import javax.servlet.ServletResponse;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.http.MediaType;
+
+import com.fasterxml.jackson.core.JsonEncoding;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import com.fasterxml.jackson.databind.ser.impl.SimpleBeanPropertyFilter;
+import com.fasterxml.jackson.databind.ser.impl.SimpleFilterProvider;
+
+import ch.psi.daq.query.model.Aggregation;
+import ch.psi.daq.query.model.QueryField;
+import ch.psi.daq.query.model.impl.DAQQuery;
+import ch.psi.daq.queryrest.response.ResponseStreamWriter;
+
+/**
+ * Takes a Java 8 stream and writes it to the output stream provided by the {@link ServletResponse}
+ * of the current request.
+ */
+public class JSONResponseStreamWriter implements ResponseStreamWriter {
+
+   private static final String DATA_RESP_FIELD = "data";
+
+   private static final Logger logger = LoggerFactory.getLogger(JSONResponseStreamWriter.class);
+
+   @Resource
+   private JsonFactory jsonFactory;
+
+   @Resource
+   private ObjectMapper mapper;
+
+   @Override
+   public void respond(Stream<Entry<String, ?>> stream, DAQQuery query, ServletResponse response) throws Exception {
+      response.setCharacterEncoding(JsonEncoding.UTF8.getJavaName());
+      response.setContentType(MediaType.APPLICATION_JSON_VALUE);
+
+      Set<String> includedFields = getFields(query);
+
+      ObjectWriter writer = configureWriter(includedFields);
+      respondInternal(stream, response, writer);
+   }
+
+   protected Set<String> getFields(DAQQuery query) {
+      Set<QueryField> queryFields = query.getFields();
+      List<Aggregation> aggregations = query.getAggregations();
+
+      Set<String> includedFields =
+            new LinkedHashSet<String>(queryFields.size() + (aggregations != null ? aggregations.size() : 0));
+
+      for (QueryField field : queryFields) {
+         includedFields.add(field.name());
+      }
+      if (aggregations != null) {
+         for (Aggregation aggregation : query.getAggregations()) {
+            includedFields.add(aggregation.name());
+         }
+      }
+
+      // do not write channel since it is already provided as key in mapping
+      includedFields.remove(QueryField.channel.name());
+
+      return includedFields;
+   }
+
+   /**
+    * Configures the writer dynamically by including the fields which should be included in the
+    * response.
+    * 
+    * @param includedFields set of strings which correspond to the getter method names of the
+    *        classes registered as a mixed-in
+    * @return the configured writer that includes the specified fields
+    */
+   private ObjectWriter configureWriter(Set<String> includedFields) {
+      SimpleFilterProvider propertyFilter = new SimpleFilterProvider();
+      propertyFilter.addFilter("namedPropertyFilter", SimpleBeanPropertyFilter.filterOutAllExcept(includedFields));
+      // only write the properties not excluded in the filter
+      ObjectWriter writer = mapper.writer(propertyFilter);
+      return writer;
+   }
+
+   /**
+    * Writes the outer Java stream into the output stream.
+    * 
+    * @param stream Mapping from channel name to data
+    * @param response {@link ServletResponse} instance given by the current HTTP request
+    * @param writer configured writer that includes the fields the end user wants to see
+    * @throws IOException thrown if writing to the output stream fails
+    */
+   private void respondInternal(Stream<Entry<String, ?>> stream, ServletResponse response, ObjectWriter writer)
+         throws Exception {
+      JsonGenerator generator = jsonFactory.createGenerator(response.getOutputStream(), JsonEncoding.UTF8);
+
+      AtomicReference<Exception> exception = new AtomicReference<>();
+      try {
+         generator.writeStartArray();
+         stream
+               /* ensure elements are sequentially written */
+               .sequential()
+               .forEach(
+                     entry -> {
+                        try {
+                           generator.writeStartObject();
+                           generator.writeStringField(QueryField.channel.name(), entry.getKey());
+
+                           generator.writeFieldName(DATA_RESP_FIELD);
+                           writer.writeValue(generator, entry.getValue());
+
+                           generator.writeEndObject();
+                        } catch (Exception e) {
+                           logger.error("Could not write channel name of channel '{}'", entry.getKey(), e);
+                           exception.compareAndSet(null, e);
+                        }
+                     });
+         generator.writeEndArray();
+      } finally {
+         generator.flush();
+         generator.close();
+      }
+
+      if (exception.get() != null) {
+         throw exception.get();
+      }
+   }
+
+}

src/main/java/ch/psi/daq/queryrest/response/json/JsonByteArraySerializer.java

@@ -1,4 +1,4 @@
-package ch.psi.daq.queryrest.response;
+package ch.psi.daq.queryrest.response.json;
 
 import java.io.IOException;
 

src/main/java/ch/psi/daq/queryrest/response/json/JsonStreamSerializer.java

@@ -1,4 +1,4 @@
-package ch.psi.daq.queryrest.response;
+package ch.psi.daq.queryrest.response.json;
 
 import java.io.IOException;
 import java.util.Iterator;