Merge branch 'dev' into v0.2.1_release

Author: blootsvoets

build.gradle

@@ -56,6 +56,6 @@ artifacts {
 }
 
 task wrapper(type: Wrapper) {
-    gradleVersion = '3.5'
+    gradleVersion = '4.1'
     distributionType 'all'
 }

gradle/wrapper/gradle-wrapper.jar

@@ -1,6 +1,5 @@
-#Mon Jul 03 09:50:31 CEST 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-3.5-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.1-all.zip

gradle/wrapper/gradle-wrapper.properties

@@ -33,11 +33,11 @@ DEFAULT_JVM_OPTS=""
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"
 
-warn ( ) {
+warn () {
     echo "$*"
 }
 
-die ( ) {
+die () {
     echo
     echo "$*"
     echo
@@ -155,7 +155,7 @@ if $cygwin ; then
 fi
 
 # Escape application args
-save ( ) {
+save () {
     for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
     echo " "
 }

gradlew

@@ -17,6 +17,7 @@
 package org.radarcns;
 
 import java.nio.file.Files;
+import java.util.Date;
 import java.util.List;
 import java.util.Objects;
 import javax.annotation.Nonnull;
@@ -63,8 +64,8 @@ public static Frequency read(File file) {
         return new Frequency(file, map);
     }
 
-    public void add(String topicName, String id, GenericRecord valueField, Field timeField) {
-        String timestamp = RestructureAvroRecords.createHourTimestamp(valueField, timeField);
+    public void add(String topicName, String id, Date date) {
+        String timestamp = RestructureAvroRecords.createHourTimestamp(date);
 
         Integer count = (Integer) bins.get(topicName, id, timestamp);
         if (count == null) {

src/main/java/org/radarcns/Frequency.java

@@ -16,44 +16,49 @@
 
 package org.radarcns;
 
-import java.util.Objects;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+
 import javax.annotation.Nonnull;
 
 /** POJO class for storing offsets. */
 public class OffsetRange implements Comparable<OffsetRange> {
-    private String topic;
-    private int partition;
+    private final String topic;
+    private final int partition;
     private long offsetFrom;
     private long offsetTo;
 
-    public static OffsetRange parse(String filename) throws NumberFormatException, IndexOutOfBoundsException {
+    public static OffsetRange parseFilename(String filename) throws NumberFormatException, IndexOutOfBoundsException {
         String[] fileNameParts = filename.split("[+.]");
 
-        OffsetRange range = new OffsetRange();
-        range.topic = fileNameParts[0];
-        range.partition = Integer.parseInt(fileNameParts[1]);
-        range.offsetFrom = Long.parseLong(fileNameParts[2]);
-        range.offsetTo = Long.parseLong(fileNameParts[3]);
-        return range;
+        return new OffsetRange(
+                fileNameParts[0],
+                Integer.parseInt(fileNameParts[1]),
+                Long.parseLong(fileNameParts[2]),
+                Long.parseLong(fileNameParts[3]));
     }
 
-    public String getTopic() {
-        return topic;
+    /** Full constructor. */
+    @JsonCreator
+    public OffsetRange(
+            @JsonProperty("topic") String topic,
+            @JsonProperty("partition") int partition,
+            @JsonProperty("offsetFrom") long offsetFrom,
+            @JsonProperty("offsetTo") long offsetTo) {
+        this.topic = topic;
+        this.partition = partition;
+        this.offsetFrom = offsetFrom;
+        this.offsetTo = offsetTo;
     }
 
-    public void setTopic(@Nonnull String topic) {
-        Objects.requireNonNull(topic);
-        this.topic = topic;
+    public String getTopic() {
+        return topic;
     }
 
     public int getPartition() {
         return partition;
     }
 
-    public void setPartition(int partition) {
-        this.partition = partition;
-    }
-
     public long getOffsetFrom() {
         return offsetFrom;
     }
@@ -97,7 +102,7 @@ public boolean equals(Object o) {
     }
 
     @Override
-    public int compareTo(OffsetRange o) {
+    public int compareTo(@Nonnull OffsetRange o) {
         int ret = Long.compare(offsetFrom, o.offsetFrom);
         if (ret != 0) {
             return ret;

src/main/java/org/radarcns/OffsetRange.java

@@ -31,60 +31,87 @@
 import java.io.FileWriter;
 import java.io.Flushable;
 import java.io.IOException;
+import java.nio.file.Files;
+
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
 
 /**
  * Accesses a OffsetRange file using the CSV format. On construction, this will create the file if
  * not present.
  */
-public class OffsetRangeFile implements Flushable, Closeable {
-    private final CsvMapper mapper;
-    private final CsvSchema schema;
-    private final File file;
-    private final FileWriter fileWriter;
-    private final BufferedWriter bufferedWriter;
-    private final CsvGenerator generator;
-    private final ObjectWriter writer;
+public final class OffsetRangeFile {
+    private static final CsvSchema SCHEMA = CsvSchema.builder()
+            .addColumn("topic")
+            .addNumberColumn("partition")
+            .addNumberColumn("offsetTo")
+            .addNumberColumn("offsetFrom")
+            .build();
+
+    private static final CsvFactory CSV_FACTORY = new CsvFactory();
+    private static final CsvMapper CSV_MAPPER = new CsvMapper(CSV_FACTORY);
+    private static final ObjectReader CSV_READER = CSV_MAPPER.reader(SCHEMA.withHeader())
+            .forType(OffsetRange.class);
 
-    public OffsetRangeFile(File file) throws IOException {
-        this.file = file;
-        boolean fileIsNew = !file.exists() || file.length() == 0;
-        CsvFactory factory = new CsvFactory();
-        this.mapper = new CsvMapper(factory);
-        this.schema = mapper.schemaFor(OffsetRange.class);
-        this.fileWriter = new FileWriter(file, true);
-        this.bufferedWriter = new BufferedWriter(this.fileWriter);
-        this.generator = factory.createGenerator(bufferedWriter);
-        this.writer = mapper.writerFor(OffsetRange.class)
-                .with(fileIsNew ? schema.withHeader() : schema);
+    private OffsetRangeFile() {
+        // utility class
     }
 
-    public void write(OffsetRange range) throws IOException {
-        writer.writeValue(generator, range);
+    public static void cleanUp(File file) throws IOException {
+        File tmpFile = File.createTempFile("offsets", ".csv.tmp");
+        try (OffsetRangeFile.Writer offsets = new OffsetRangeFile.Writer(tmpFile)) {
+            offsets.write(OffsetRangeFile.read(file));
+        }
+        Files.move(tmpFile.toPath(), file.toPath(), REPLACE_EXISTING);
     }
 
-    public OffsetRangeSet read() throws IOException {
+    public static OffsetRangeSet read(File inputFile) throws IOException {
         OffsetRangeSet set = new OffsetRangeSet();
-        ObjectReader reader = mapper.readerFor(OffsetRange.class).with(schema.withHeader());
 
-        try (FileReader fr = new FileReader(file);
+        try (FileReader fr = new FileReader(inputFile);
                 BufferedReader br = new BufferedReader(fr)) {
-            MappingIterator<OffsetRange> ranges = reader.readValues(br);
+            MappingIterator<OffsetRange> ranges = CSV_READER.readValues(br);
             while(ranges.hasNext()) {
                 set.add(ranges.next());
             }
         }
         return set;
     }
 
-    @Override
-    public void flush() throws IOException {
-        generator.flush();
-    }
+    public static class Writer implements Flushable, Closeable {
+        private final FileWriter fileWriter;
+        private final BufferedWriter bufferedWriter;
+        private final CsvGenerator generator;
+        private final ObjectWriter writer;
+
+        public Writer(File outputFile) throws IOException {
+            boolean fileIsNew = !outputFile.exists() || outputFile.length() == 0;
+            this.fileWriter = new FileWriter(outputFile, true);
+            this.bufferedWriter = new BufferedWriter(this.fileWriter);
+            this.generator = CSV_FACTORY.createGenerator(bufferedWriter);
+            this.writer = CSV_MAPPER.writerFor(OffsetRange.class)
+                    .with(fileIsNew ? SCHEMA.withHeader() : SCHEMA);
+        }
+
+        public void write(OffsetRange range) throws IOException {
+            writer.writeValue(generator, range);
+        }
+
+        public void write(OffsetRangeSet rangeSet) throws IOException {
+            for (OffsetRange range : rangeSet) {
+                write(range);
+            }
+        }
 
-    @Override
-    public void close() throws IOException {
-        generator.close();
-        bufferedWriter.close();
-        fileWriter.close();
+        @Override
+        public void flush() throws IOException {
+            generator.flush();
+        }
+
+        @Override
+        public void close() throws IOException {
+            generator.close();
+            bufferedWriter.close();
+            fileWriter.close();
+        }
     }
 }