Merge remote-tracking branch 'origin/dev' into release-0.3.2

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

@@ -312,7 +312,7 @@ protected Duration getLookbackTime() {
  */
  protected Instant nextPoll(User user) {
  Instant offset = getOffset(user);
- if (offset.isAfter(user.getEndDate())) {
+ if (offset.isAfter(user.getEndDate().minus(getEndDateThreshold()))) {
  return nearFuture();
  } else {
  Instant nextPoll = lastPollPerUser.getOrDefault(user.getId(), MIN_INSTANT)
@@ -321,6 +321,10 @@ protected Instant nextPoll(User user) {
  }
  }
 
+ private TemporalAmount getEndDateThreshold() {
+ return Duration.ofHours(1);
+ }
+
  /**
  * Generate one date per day, using UTC time zone. The first date will have the time from the
  * given startDate. Following time stamps will start at 00:00. This will not up to the date of

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

@@ -45,6 +45,9 @@ public class LocalUser implements User {
  @JsonProperty("oauth2")
  private OAuth2UserCredentials oauth2Credentials = new OAuth2UserCredentials();
 
+ @JsonProperty("authorized")
+ private Boolean isAuthorized;
+
  @JsonIgnore
  private SchemaAndValue observationKey;
 
@@ -95,6 +98,15 @@ public void setFitbitUserId(String id) {
  this.externalUserId = id;
  }
 
+ @Override
+ public boolean isAuthorized() {
+ if(isAuthorized == null) {
+ return !oauth2Credentials.isAccessTokenExpired()
+ || oauth2Credentials.hasRefreshToken();
+ }
+ return isAuthorized;
+ }
+
  @Override
  public String getVersion() {
  return version;
@@ -115,6 +127,7 @@ public LocalUser copy() {
  copy.endDate = endDate;
  copy.sourceId = sourceId;
  copy.oauth2Credentials = oauth2Credentials;
+ copy.isAuthorized = isAuthorized;
  return copy;
  }
 
@@ -133,6 +146,7 @@ public String toString() {
  + ", projectId='" + projectId + '\''
  + ", userId='" + userId + '\''
  + ", sourceId='" + sourceId + '\''
+ + ", isAuthorized='" + isAuthorized() + '\''
  + '}';
  }
 
@@ -152,7 +166,8 @@ public boolean equals(Object o) {
  && Objects.equals(userId, localUser.userId)
  && Objects.equals(sourceId, localUser.sourceId)
  && Objects.equals(startDate, localUser.startDate)
- && Objects.equals(endDate, localUser.endDate);
+ && Objects.equals(endDate, localUser.endDate)
+ && Objects.equals(isAuthorized(), localUser.isAuthorized());
  }
 
  @Override

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

@@ -45,6 +45,8 @@ static SchemaAndValue computeObservationKey(AvroData avroData, User user) {
 
  String getSourceId();
 
+ boolean isAuthorized();
+
  default String getVersionedId() {
  String version = getVersion();
  if (version == null) {
@@ -60,6 +62,7 @@ default Boolean isComplete() {
  return getEndDate() != null
  && getStartDate() != null
  && getProjectId() != null
- && getUserId() != null;
+ && getUserId() != null
+ && isAuthorized();
  }
 }

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

@@ -70,6 +70,12 @@ public String getSourceId() {
  return fitbitAuthDetails.getSourceId();
  }
 
+ @Override
+ public boolean isAuthorized() {
+ return !fitbitAuthDetails.getOauth2Credentials().isAccessTokenExpired()
+ || fitbitAuthDetails.getOauth2Credentials().hasRefreshToken();
+ }
+
  public FirebaseUserDetails getFirebaseUserDetails() {
  return firebaseUserDetails;
  }