Add OuraUserRepository abstract class and extend from this

Author: mpgxvii

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

@@ -43,8 +43,8 @@
 import org.apache.kafka.common.config.ConfigDef.Width;
 import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.connect.errors.ConnectException;
+import org.radarbase.connect.rest.oura.user.OuraUserRepository;
 import org.radarbase.connect.rest.oura.user.OuraServiceUserRepository;
-import org.radarbase.connect.rest.oura.user.OuraServiceUserRepositoryLegacy;
 
 public class OuraRestSourceConnectorConfig extends AbstractConfig {
   public static final Pattern COLON_PATTERN = Pattern.compile(":");
@@ -102,7 +102,7 @@ public class OuraRestSourceConnectorConfig extends AbstractConfig {
   private static final String OURA_USER_REPOSITORY_TOKEN_URL_DOC = "OAuth 2.0 token url for retrieving client credentials.";
   private static final String OURA_USER_REPOSITORY_TOKEN_URL_DISPLAY = "OAuth 2.0 token URL.";
 
-  private OuraServiceUserRepository userRepository;
+  private OuraUserRepository userRepository;
   private final Headers clientCredentials;
 
   public OuraRestSourceConnectorConfig(ConfigDef config, Map<String, String> parsedConfig, boolean doLog) {
@@ -193,7 +193,7 @@ public String toString() {
 
         .define(OURA_USER_REPOSITORY_CONFIG,
             Type.CLASS,
-            OuraServiceUserRepositoryLegacy.class,
+            OuraServiceUserRepository.class,
             Importance.MEDIUM,
             OURA_USER_REPOSITORY_DOC,
             group,
@@ -255,7 +255,7 @@ public String getOuraClientSecret() {
     return getPassword(OURA_API_SECRET_CONFIG).value();
   }
 
-  public OuraServiceUserRepository getUserRepository(OuraServiceUserRepository reuse) {
+  public OuraUserRepository getUserRepository(OuraUserRepository reuse) {
     if (reuse != null && reuse.getClass().equals(getClass(OURA_USER_REPOSITORY_CONFIG))) {
       userRepository = reuse;
     } else {
@@ -265,15 +265,15 @@ public OuraServiceUserRepository getUserRepository(OuraServiceUserRepository reu
     return userRepository;
   }
 
-  public OuraServiceUserRepository getUserRepository() {
+  public OuraUserRepository getUserRepository() {
     userRepository.initialize(this);
     return userRepository;
   }
 
   @SuppressWarnings("unchecked")
-  public OuraServiceUserRepository createUserRepository() {
+  public OuraUserRepository createUserRepository() {
     try {
-      return ((Class<? extends OuraServiceUserRepository>)
+      return ((Class<? extends OuraUserRepository>)
           getClass(OURA_USER_REPOSITORY_CONFIG)).getDeclaredConstructor().newInstance();
     } catch (IllegalAccessException | InstantiationException
         | InvocationTargetException | NoSuchMethodException e) {

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

@@ -29,7 +29,7 @@
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
 import org.radarbase.oura.user.User;
-import org.radarbase.connect.rest.oura.user.OuraServiceUserRepository;
+import org.radarbase.connect.rest.oura.user.OuraUserRepository;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -45,7 +45,7 @@ public class OuraSourceConnector extends AbstractRestSourceConnector {
   private static final Logger logger = LoggerFactory.getLogger(OuraSourceConnector.class);
   private ScheduledExecutorService executor;
   private Set<? extends User> configuredUsers;
-  private OuraServiceUserRepository repository;
+  private OuraUserRepository repository;
 
   @Override
   public void start(Map<String, String> props) {

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

@@ -35,7 +35,7 @@
 import org.apache.kafka.connect.source.SourceTask;
 import org.apache.kafka.connect.storage.OffsetStorageReader;
 import org.radarbase.connect.rest.oura.offset.KafkaOffsetManager;
-import org.radarbase.connect.rest.oura.user.OuraServiceUserRepository;
+import org.radarbase.connect.rest.oura.user.OuraUserRepository;
 import org.radarbase.connect.rest.oura.util.VersionUtil;
 import org.radarbase.oura.converter.TopicData;
 import org.radarbase.oura.request.OuraRequestGenerator;
@@ -57,7 +57,7 @@ public class OuraSourceTask extends SourceTask {
   private static final Logger logger = LoggerFactory.getLogger(OuraSourceTask.class);
 
   private OkHttpClient baseClient;
-  private OuraServiceUserRepository userRepository;
+  private OuraUserRepository userRepository;
   private List<Route> routes;
   private OuraRequestGenerator ouraRequestGenerator;
   private AvroData avroData = new AvroData(20);

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraServiceUserRepository.kt

@@ -67,7 +67,7 @@ import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 
 @Suppress("unused")
-class OuraServiceUserRepository : OuraServiceUserRepositoryLegacy() {
+class OuraServiceUserRepository : OuraUserRepository() {
     private lateinit var userCache: CachedSet<OuraUser>
     private lateinit var client: HttpClient
     private val credentialCaches = ConcurrentHashMap<String, CachedValue<OAuth2UserCredentials>>()

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraServiceUserRepositoryLegacy.java

@@ -60,14 +60,14 @@
  import static kotlin.sequences.SequencesKt.*;
 
  @SuppressWarnings("unused")
- public class OuraServiceUserRepositoryLegacy implements UserRepository {
+ public class OuraServiceUserRepositoryLegacy extends OuraUserRepository {
   Instant MIN_INSTANT = Instant.EPOCH;
 
   public static final JsonFactory JSON_FACTORY = new JsonFactory();
    public static final ObjectReader JSON_READER = new ObjectMapper(JSON_FACTORY)
        .registerModule(new JavaTimeModule())
        .reader();
-   private static final Logger logger = LoggerFactory.getLogger(OuraServiceUserRepository.class);
+   private static final Logger logger = LoggerFactory.getLogger(OuraServiceUserRepositoryLegacy.class);
 
    private static final ObjectReader USER_LIST_READER = JSON_READER.forType(OuraUsers.class);
    private static final ObjectReader USER_READER = JSON_READER.forType(User.class);
@@ -153,6 +153,7 @@ public String getAccessToken(User user) throws IOException, UserNotAuthorizedExc
      }
    }
 
+   @Override
    public String refreshAccessToken(User user) throws IOException, UserNotAuthorizedException {
      if (!user.isAuthorized()) {
        throw new UserNotAuthorizedException("User is not authorized");

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/OuraUserRepository.kt

@@ -0,0 +1,41 @@
+/*
+ * Copyright 2018 The Hyve
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+package org.radarbase.connect.rest.oura.user
+
+import org.radarbase.connect.rest.oura.OuraRestSourceConnectorConfig
+import org.radarbase.oura.user.User
+import org.radarbase.oura.user.UserNotAuthorizedException
+import org.radarbase.oura.user.UserRepository
+import org.slf4j.LoggerFactory
+import java.io.IOException
+
+@Suppress("unused")
+abstract class OuraUserRepository : UserRepository {
+    abstract fun initialize(config: OuraRestSourceConnectorConfig)
+
+    @Throws(IOException::class, UserNotAuthorizedException::class)
+    abstract fun refreshAccessToken(user: User): String
+
+    @Throws(IOException::class)
+    abstract fun applyPendingUpdates()
+
+    abstract fun hasPendingUpdates(): Boolean
+
+    companion object {
+        private val logger = LoggerFactory.getLogger(OuraUserRepository::class.java)
+    }
+}

kafka-connect-oura-source/src/main/java/org/radarbase/connect/rest/oura/user/UserNotAuthorizedException.java

@@ -17,7 +17,7 @@
 
 package org.radarbase.connect.rest.oura.user;
 
-public class UserNotAuthorizedException extends Exception {
+public class UserNotAuthorizedException extends RuntimeException {
     public UserNotAuthorizedException(String message) {
         super(message);
     }