Merge pull request #65 from RADAR-base/filterAfterEndDate

Filter after end date

Author: blootsvoets

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/converter/FitbitAvroConverter.java

@@ -27,18 +27,15 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.Iterator;
 import java.util.Map;
-import java.util.Objects;
 import java.util.Optional;
-import java.util.OptionalDouble;
-import java.util.OptionalLong;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 import okhttp3.Response;
 import okhttp3.ResponseBody;
+import org.apache.avro.Schema.Field;
 import org.apache.avro.generic.IndexedRecord;
 import org.apache.kafka.connect.data.SchemaAndValue;
 import org.apache.kafka.connect.source.SourceRecord;
@@ -86,7 +83,7 @@ public Collection<SourceRecord> convert(
     double timeReceived = System.currentTimeMillis() / 1000d;
 
     return processRecords((FitbitRestRequest)restRequest, activities, timeReceived)
-        .filter(Objects::nonNull)
+        .filter(t -> validateRecord((FitbitRestRequest)restRequest, t))
         .map(t -> {
           SchemaAndValue avro = avroData.toConnectData(t.value.getSchema(), t.value);
           Map<String, ?> offset = Collections.singletonMap(
@@ -98,6 +95,22 @@ public Collection<SourceRecord> convert(
         .collect(Collectors.toList());
   }
 
+  private boolean validateRecord(FitbitRestRequest request, TopicData record) {
+    if (record == null) {
+      return false;
+    }
+    Instant endDate = request.getUser().getEndDate();
+    if (endDate == null) {
+      return true;
+    }
+    Field timeField = record.value.getSchema().getField("time");
+    if (timeField != null) {
+      long time = (long) (((Double)record.value.get(timeField.pos()) * 1000.0));
+      return Instant.ofEpochMilli(time).isBefore(endDate);
+    }
+    return true;
+  }
+
   /** Process the JSON records generated by given request. */
   protected abstract Stream<TopicData> processRecords(
       FitbitRestRequest request,