Merge pull request #62 from RADAR-base/close_requests

Close requests

Author: blootsvoets

build.gradle

@@ -11,7 +11,7 @@ subprojects {
     apply plugin: 'java-library'
 
     group = 'org.radarbase'
-    version = '0.3.1'
+    version = '0.3.2-SNAPSHOT'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8

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

@@ -209,6 +209,61 @@ public synchronized String refreshAccessToken(User user, int retry) throws IOExc
       throw new NoSuchElementException("User " + user + " is not present in this user repository.");
     }
     String refreshToken = actualUser.apply(u -> u.getOAuth2Credentials().getRefreshToken());
+
+    JsonNode node;
+
+    try {
+      node = requestAccessToken(refreshToken);
+    } catch (IOException ex) {
+      if (retry > 0) {
+        logger.warn("Failed to read OAuth 2.0 response: {}", ex.toString());
+        return refreshAccessToken(user, retry - 1);
+      } else {
+        throw ex;
+      }
+    } catch (NotAuthorizedException ex) {
+      actualUser.accept((u, p) -> {
+        if (!refreshToken.equals(u.getOAuth2Credentials().getRefreshToken())) {
+          // it was updated already by another thread.
+          return;
+        }
+        u.setOauth2Credentials(new OAuth2UserCredentials());
+        store(p, u);
+      });
+      throw ex;
+    }
+
+    JsonNode expiresInNode = node.get("expires_in");
+    Long expiresIn = expiresInNode != null
+        ? expiresInNode.asLong()
+        : null;
+
+    JsonNode accessTokenNode = node.get("access_token");
+    JsonNode refreshTokenNode = node.get("refresh_token");
+    if (accessTokenNode == null || refreshTokenNode == null) {
+      if (retry > 0) {
+        logger.warn("Failed to get access token in successful OAuth 2.0 request:"
+            + " access token or refresh token are missing");
+        return refreshAccessToken(user, retry - 1);
+      } else {
+        throw new NotAuthorizedException("Did not get an access token");
+      }
+    }
+
+    actualUser.accept((u, p) -> {
+      if (!refreshToken.equals(u.getOAuth2Credentials().getRefreshToken())) {
+        // it was updated already by another thread.
+        return;
+      }
+      u.setOauth2Credentials(new OAuth2UserCredentials(
+          refreshTokenNode.asText(), accessTokenNode.asText(), expiresIn));
+      store(p, u);
+    });
+
+    return actualUser.apply(u -> u.getOAuth2Credentials().getAccessToken());
+  }
+
+  private JsonNode requestAccessToken(String refreshToken) throws IOException {
     if (refreshToken == null || refreshToken.isEmpty()) {
       throw new NotAuthorizedException("Refresh token is not set");
     }
@@ -225,52 +280,8 @@ public synchronized String refreshAccessToken(User user, int retry) throws IOExc
       ResponseBody responseBody = response.body();
 
       if (response.isSuccessful() && responseBody != null) {
-        JsonNode node;
-        try {
-          node = JSON_READER.readTree(responseBody.charStream());
-        } catch (IOException ex) {
-          if (retry > 0) {
-            logger.warn("Failed to read OAuth 2.0 response: {}", ex.toString());
-            return refreshAccessToken(user, retry - 1);
-          }
-          throw ex;
-        }
-
-        JsonNode expiresInNode = node.get("expires_in");
-        Long expiresIn = expiresInNode != null
-            ? expiresInNode.asLong()
-            : null;
-
-        JsonNode accessTokenNode = node.get("access_token");
-        JsonNode refreshTokenNode = node.get("refresh_token");
-        if (accessTokenNode == null || refreshTokenNode == null) {
-          if (retry > 0) {
-            logger.warn("Failed to get access token in successful OAuth 2.0 request:"
-                + " access token or refresh token are missing");
-            return refreshAccessToken(user, retry - 1);
-          } else {
-            throw new NotAuthorizedException("Did not get an access token");
-          }
-        }
-
-        actualUser.accept((u, p) -> {
-          if (!refreshToken.equals(u.getOAuth2Credentials().getRefreshToken())) {
-            // it was updated already by another thread.
-            return;
-          }
-          u.setOauth2Credentials(new OAuth2UserCredentials(
-              refreshTokenNode.asText(), accessTokenNode.asText(), expiresIn));
-          store(p, u);
-        });
+        return JSON_READER.readTree(responseBody.charStream());
       } else if (response.code() == 400 || response.code() == 401) {
-        actualUser.accept((u, p) -> {
-          if (!refreshToken.equals(u.getOAuth2Credentials().getRefreshToken())) {
-            // it was updated already by another thread.
-            return;
-          }
-          u.setOauth2Credentials(new OAuth2UserCredentials());
-          store(p, u);
-        });
         throw new NotAuthorizedException("Refresh token is no longer valid.");
       } else {
         String message = "Failed to request refresh token, with response HTTP status code "
@@ -281,7 +292,6 @@ public synchronized String refreshAccessToken(User user, int retry) throws IOExc
         throw new IOException(message);
       }
     }
-    return actualUser.apply(u -> u.getOAuth2Credentials().getAccessToken());
   }
 
   /**

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

@@ -80,22 +80,25 @@ public Stream<SourceRecord> handleRequest() throws IOException {
       return Stream.empty();
     }
 
+    Collection<SourceRecord> records;
+
     try (Response response = client.newCall(request).execute()) {
       if (!response.isSuccessful()) {
         route.requestFailed(this, response);
         return Stream.empty();
       }
 
-      Collection<SourceRecord> records = route.converter().convert(this, response);
-      if (records.isEmpty()) {
-        route.requestEmpty(this);
-      } else {
-        records.forEach(r -> route.requestSucceeded(this, r));
-      }
-      return records.stream();
+      records = route.converter().convert(this, response);
     } catch (IOException ex) {
       route.requestFailed(this, null);
       throw ex;
     }
+
+    if (records.isEmpty()) {
+      route.requestEmpty(this);
+    } else {
+      records.forEach(r -> route.requestSucceeded(this, r));
+    }
+    return records.stream();
   }
 }