Merge pull request #40 from RADAR-base/fix-npe

Fix NPE

Author: blootsvoets

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/ServiceUserRepository.java

@@ -30,13 +30,10 @@
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.Set;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import javax.ws.rs.NotAuthorizedException;
-
 import okhttp3.HttpUrl;
 import okhttp3.MediaType;
 import okhttp3.OkHttpClient;
@@ -56,15 +53,17 @@ public class ServiceUserRepository implements UserRepository {
   private static final ObjectReader USER_LIST_READER = JSON_READER.forType(Users.class);
   private static final ObjectReader USER_READER = JSON_READER.forType(User.class);
   private static final ObjectReader OAUTH_READER = JSON_READER.forType(OAuth2UserCredentials.class);
-  private static final RequestBody EMPTY_BODY = RequestBody.create(
-      MediaType.parse("application/json; charset=utf-8"), "");
+  private static final RequestBody EMPTY_BODY =
+      RequestBody.create(MediaType.parse("application/json; charset=utf-8"), "");
+  private static final Duration FETCH_THRESHOLD = Duration.ofMinutes(1L);
+
   private final OkHttpClient client;
-  private HttpUrl baseUrl;
   private final Map<String, OAuth2UserCredentials> cachedCredentials;
-  private HashSet<String> containedUsers;
-  private Set< ? extends User> timedCachedUsers = new HashSet<>();
   private final AtomicReference<Instant> nextFetch = new AtomicReference<>(MIN_INSTANT);
-  private static final Duration FETCH_THRESHOLD = Duration.ofMinutes(1L);
+
+  private HttpUrl baseUrl;
+  private HashSet<String> containedUsers;
+  private Set<? extends User> timedCachedUsers = new HashSet<>();
 
   public ServiceUserRepository() {
     this.client = new OkHttpClient();
@@ -97,10 +96,11 @@ public Stream<? extends User> stream() throws IOException {
 
     logger.info("Requesting user information from webservice");
     Request request = requestFor("users" + "?source-type=FitBit").build();
-    this.timedCachedUsers = this.<Users>makeRequest(request, USER_LIST_READER)
-        .getUsers().stream()
-        .filter(u -> containedUsers.isEmpty() || containedUsers.contains(u.getId()))
-        .collect(Collectors.toSet());
+    this.timedCachedUsers =
+        this.<Users>makeRequest(request, USER_LIST_READER).getUsers().stream()
+            .filter(u -> u.isComplete()
+                && (containedUsers.isEmpty() || containedUsers.contains(u.getId())))
+            .collect(Collectors.toSet());
 
     return this.timedCachedUsers.stream();
   }
@@ -118,9 +118,7 @@ public String getAccessToken(User user) throws IOException, NotAuthorizedExcepti
 
   @Override
   public String refreshAccessToken(User user) throws IOException, NotAuthorizedException {
-    Request request = requestFor("users/" + user.getId() + "/token")
-      .post(EMPTY_BODY)
-      .build();
+    Request request = requestFor("users/" + user.getId() + "/token").post(EMPTY_BODY).build();
     OAuth2UserCredentials credentials = makeRequest(request, OAUTH_READER);
     cachedCredentials.put(user.getId(), credentials);
     return credentials.getAccessToken();

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/user/User.java

@@ -23,18 +23,32 @@
 import org.radarcns.kafka.ObservationKey;
 
 public interface User {
+  static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
+    return avroData.toConnectData(
+        ObservationKey.getClassSchema(),
+        new ObservationKey(user.getProjectId(), user.getUserId(), user.getSourceId()));
+  }
+
   String getId();
+
   String getExternalUserId();
+
   String getProjectId();
+
   String getUserId();
+
   Instant getStartDate();
+
   Instant getEndDate();
+
   String getSourceId();
+
   SchemaAndValue getObservationKey(AvroData avroData);
 
-  static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
-    return avroData.toConnectData(
-          ObservationKey.getClassSchema(),
-          new ObservationKey(user.getProjectId(), user.getUserId(), user.getSourceId()));
+  default Boolean isComplete() {
+    return getEndDate() != null
+        && getStartDate() != null
+        && getProjectId() != null
+        && getUserId() != null;
   }
 }