Use iterator instead of stream to ensure once-only requests

Author: blootsvoets

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitIntradayHeartRateRoute.java

@@ -29,12 +29,11 @@
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 
 public class FitbitIntradayHeartRateRoute extends FitbitPollingRoute {
-  private static final String ROUTE_NAME = "heart_rate";
   private final FitbitIntradayHeartRateAvroConverter converter;
 
   public FitbitIntradayHeartRateRoute(FitbitRequestGenerator generator,
       UserRepository userRepository, AvroData avroData) {
-    super(generator, userRepository, ROUTE_NAME);
+    super(generator, userRepository, "heart_rate");
     this.converter = new FitbitIntradayHeartRateAvroConverter(avroData);
   }
 

kafka-connect-rest-source/src/main/java/org/radarbase/connect/rest/RestSourceTask.java

@@ -20,10 +20,12 @@
 import static java.time.temporal.ChronoUnit.MILLIS;
 import static org.radarbase.connect.rest.util.ThrowingFunction.tryOrNull;
 
+import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.time.Instant;
 import java.util.ArrayList;
 import java.util.Collections;
+import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -63,8 +65,8 @@ public void start(Map<String, String> map) {
 
   @Override
   public List<SourceRecord> poll() throws InterruptedException {
-    LongAdder requestsGenerated = new LongAdder();
-    List<SourceRecord> requests;
+    long requestsGenerated = 0;
+    List<SourceRecord> requests = Collections.emptyList();
 
     do {
       long timeout = MILLIS.between(Instant.now(), requestGenerator.getTimeOfNextRequest());
@@ -73,23 +75,27 @@ public List<SourceRecord> poll() throws InterruptedException {
         Thread.sleep(timeout);
       }
 
-      requests = requestGenerator.requests()
-          .sequential()
+      Iterator<? extends RestRequest> requestIterator = requestGenerator.requests()
           .filter(RestRequest::isStillValid)
-          .peek(r -> {
-            logger.info("Requesting {}", r.getRequest().url());
-            requestsGenerated.increment();
-          })
-          .map(tryOrNull(RestRequest::handleRequest,
-              (r, ex) -> logger.warn("Failed to make request: {}", ex.toString())))
-          .filter(Objects::nonNull)
-          .map(s -> s.collect(Collectors.toList()))
-          .filter(l -> !l.isEmpty())
-          .findFirst()
-          .orElse(Collections.emptyList());
+          .iterator();
+
+
+      while (requests.isEmpty() && requestIterator.hasNext()) {
+        RestRequest request = requestIterator.next();
+
+        logger.info("Requesting {}", request.getRequest().url());
+        requestsGenerated++;
+
+        try {
+          requests = request.handleRequest()
+              .collect(Collectors.toList());
+        } catch (IOException ex) {
+          logger.warn("Failed to make request: {}", ex.toString());
+        }
+      }
     } while (requests.isEmpty());
 
-    logger.info("Processed {} records from {} URLs", requests.size(), requestsGenerated.sum());
+    logger.info("Processed {} records from {} URLs", requests.size(), requestsGenerated);
 
     return requests;
   }