More work in progress

Author: Crim

src/main/java/org/sourcelab/kafka/connect/apiclient/ApiClient.java

@@ -2,8 +2,25 @@
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.sourcelab.kafka.connect.apiclient.request.JacksonFactory;
+import org.sourcelab.kafka.connect.apiclient.request.Request;
+import org.sourcelab.kafka.connect.apiclient.request.RequestErrorResponse;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorStatus;
+import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnector;
+import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectorConfig;
+import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectorStatus;
+import org.sourcelab.kafka.connect.apiclient.request.get.connector.GetConnectors;
+import org.sourcelab.kafka.connect.apiclient.request.dto.ConnectorDefinition;
+import org.sourcelab.kafka.connect.apiclient.request.post.connector.PostConnectors;
+import org.sourcelab.kafka.connect.apiclient.request.put.connector.PutConnectorConfig;
 import org.sourcelab.kafka.connect.apiclient.rest.HttpClientRestClient;
+import org.sourcelab.kafka.connect.apiclient.rest.InvalidRequestException;
 import org.sourcelab.kafka.connect.apiclient.rest.RestClient;
+import org.sourcelab.kafka.connect.apiclient.rest.RestResponse;
+
+import java.io.IOException;
+import java.util.Collection;
+import java.util.Map;
 
 public class ApiClient {
     private static final Logger logger = LoggerFactory.getLogger(ApiClient.class);
@@ -18,6 +35,12 @@ public class ApiClient {
      */
     private final RestClient restClient;
 
+    /**
+     * Internal State flag.
+     */
+    private boolean isInitialized = false;
+
+
     /**
      * Default Constructor.
      * @param configuration Api Client Configuration.
@@ -37,4 +60,86 @@ public ApiClient(final Configuration configuration, final RestClient restClient)
         this.configuration = configuration;
         this.restClient = restClient;
     }
+
+    private <T> T submitRequest(final Request<T> request) {
+        // Submit request
+        final RestResponse restResponse = getRestClient().submitRequest(request);
+        final int responseCode = restResponse.getHttpCode();
+        String responseStr = restResponse.getResponseStr();
+
+        // If we have a valid response
+        logger.info("Response: {}", restResponse);
+
+        // Check for invalid http status codes
+        if (responseCode >= 200 && responseCode < 300) {
+            // These response codes have no values
+            if ((responseCode == 204 || responseCode == 205) && responseStr == null) {
+                // Avoid NPE
+                responseStr = "";
+            }
+
+            try {
+                return request.parseResponse(restResponse.getResponseStr());
+            } catch (IOException exception) {
+                throw new RuntimeException(exception.getMessage(), exception);
+            }
+        }
+
+        // Attempt to parse error response
+        try {
+            final RequestErrorResponse errorResponse = JacksonFactory.newInstance().readValue(responseStr, RequestErrorResponse.class);
+            throw new InvalidRequestException(errorResponse.getMessage(), errorResponse.getErrorCode());
+        } catch (IOException e) {
+            // swallow
+        }
+        throw new InvalidRequestException("Invalid response from server: " + responseStr, restResponse.getHttpCode());
+
+    }
+
+    public Collection<String> getConnectors() {
+        return submitRequest(new GetConnectors());
+    }
+
+    public ConnectorDefinition getConnector(final String connectorName) {
+        return submitRequest(new GetConnector(connectorName));
+    }
+
+    public Map<String, String> getConnectorConfig(final String connectorName) {
+        return submitRequest(new GetConnectorConfig(connectorName));
+    }
+
+    public ConnectorStatus getConnectorStatus(final String connectorName) {
+        return submitRequest(new GetConnectorStatus(connectorName));
+    }
+
+    // TODO Add return value
+    public String addConnector(final ConnectorDefinition connectorDefinition) {
+        return submitRequest(new PostConnectors(connectorDefinition));
+    }
+
+    public ConnectorDefinition updateConnectorConfig(final String connectorName, final Map<String, String> config) {
+        return submitRequest(new PutConnectorConfig(connectorName, config));
+    }
+
+    /**
+     * package protected for access in tests.
+     * @return Rest Client.
+     */
+    private RestClient getRestClient() {
+        // If we haven't initialized.
+        if (!isInitialized) {
+            // Call Init.
+            restClient.init(getConfiguration());
+
+            // Flip state flag
+            isInitialized = true;
+        }
+
+        // return our rest client.
+        return restClient;
+    }
+
+    private Configuration getConfiguration() {
+        return configuration;
+    }
 }

src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java

@@ -1,36 +1,13 @@
-/**
- * Copyright 2017 Stephen Powis https://github.com/Crim/pardot-java-client
- *
- * Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated
- * documentation files (the "Software"), to deal in the Software without restriction, including without limitation the
- * rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit
- * persons to whom the Software is furnished to do so, subject to the following conditions:
- *
- * The above copyright notice and this permission notice shall be included in all copies or substantial portions of the
- * Software.
- *
- * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE
- * WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
- * COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR
- * OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
- */
-
 package org.sourcelab.kafka.connect.apiclient;
 
 /**
- * Configure your Pardot API credentials.
+ * Configure your Kafka Connect API client.
  *
  * Also allows for configuring an optional proxy with or without authentication.
  */
 public class Configuration {
-    private final String email;
-    private final String password;
-    private final String userKey;
-
-    /**
-     * Optionally you can re-use an existing known good api key.
-     */
-    private String apiKey = null;
+    // Defines the URL/Hostname of Kafka-Connect
+    private final String apiHost;
 
     // Optional Proxy Configuration
     private String proxyHost = null;
@@ -41,32 +18,21 @@ public class Configuration {
     private String proxyUsername = null;
     private String proxyPassword = null;
 
-    // If you want to override the Pardot API url or version.
-    private String pardotApiHost = "https://pi.pardot.com/api";
-    private String pardotApiVersion = "3";
-
     /**
-     * Constructor.
-     * @param email Pardot user's email address.
-     * @param password Pardot user's password.
-     * @param userKey Pardot user's userKey.
+     * Default Constructor.
+     * @param kafkaConnectHost Hostname of Kafka-Connect
      */
-    public Configuration(final String email, final String password, final String userKey) {
-        this.email = email;
-        this.password = password;
-        this.userKey = userKey;
-    }
-
-    public String getEmail() {
-        return email;
-    }
-
-    public String getPassword() {
-        return password;
-    }
+    public Configuration(final String kafkaConnectHost) {
+        if (kafkaConnectHost == null) {
+            throw new NullPointerException("Kafka Connect Host parameter cannot be null!");
+        }
 
-    public String getUserKey() {
-        return userKey;
+        // Normalize into "http://<hostname>"
+        if (!kafkaConnectHost.startsWith("http://")) {
+            this.apiHost = "http://" + kafkaConnectHost;
+        } else {
+            this.apiHost = kafkaConnectHost;
+        }
     }
 
     /**
@@ -97,24 +63,6 @@ public Configuration useProxyAuthentication(final String proxyUsername, final St
         return this;
     }
 
-    /**
-     * Configure library to use Pardot Api Version 4.
-     * @return Configuration instance.
-     */
-    public Configuration withApiVersion4() {
-        this.pardotApiVersion = "4";
-        return this;
-    }
-
-    /**
-     * Configure library to use Pardot Api Version 4.
-     * @return Configuration instance.
-     */
-    public Configuration withApiVersion3() {
-        this.pardotApiVersion = "3";
-        return this;
-    }
-
     public String getProxyHost() {
         return proxyHost;
     }
@@ -135,37 +83,14 @@ public String getProxyPassword() {
         return proxyPassword;
     }
 
-    public String getPardotApiHost() {
-        return pardotApiHost;
-    }
-
-    public void setPardotApiHost(final String pardotApiHost) {
-        this.pardotApiHost = pardotApiHost;
-    }
-
-    public String getPardotApiVersion() {
-        return pardotApiVersion;
-    }
-
-    public void setPardotApiVersion(final String pardotApiVersion) {
-        this.pardotApiVersion = pardotApiVersion;
-    }
-
-    public String getApiKey() {
-        return apiKey;
-    }
-
-    public void setApiKey(final String apiKey) {
-        this.apiKey = apiKey;
+    public String getApiHost() {
+        return apiHost;
     }
 
     @Override
     public String toString() {
         final StringBuilder stringBuilder = new StringBuilder("Configuration{")
-            .append("email='").append(email).append('\'')
-            .append(", password='XXXXX'")
-            .append(", userKey='").append(userKey.substring(0,3)).append("...'");
-
+            .append("apiHost='").append(apiHost).append('\'');
         if (proxyHost != null) {
             stringBuilder
                 .append(", proxy='").append(proxyScheme).append("://");

src/main/java/org/sourcelab/kafka/connect/apiclient/request/JacksonFactory.java

@@ -0,0 +1,32 @@
+package org.sourcelab.kafka.connect.apiclient.request;
+
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.PropertyNamingStrategy;
+import com.fasterxml.jackson.dataformat.xml.JacksonXmlModule;
+import com.fasterxml.jackson.dataformat.xml.XmlMapper;
+import com.fasterxml.jackson.datatype.joda.JodaModule;
+
+import java.text.SimpleDateFormat;
+
+/**
+ * Creates properly configured Jackson XML Mapper instances.
+ */
+public class JacksonFactory {
+
+    /**
+     * Creates properly configured Jackson Object Mapper instances.
+     * @return ObjectMapper instance.
+     */
+    public static ObjectMapper newInstance() {
+        // Create new mapper
+        final ObjectMapper mapper = new ObjectMapper();
+
+        // Configure it
+        mapper
+            .configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false)
+            .setPropertyNamingStrategy(PropertyNamingStrategy.SNAKE_CASE);
+
+        return mapper;
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/Request.java

@@ -1,19 +1,28 @@
 package org.sourcelab.kafka.connect.apiclient.request;
 
+import java.io.IOException;
+import java.util.Collection;
 import java.util.Map;
 
 /**
  * Interface for all Requests to implement.
  */
-public interface Request {
+public interface Request<T> {
 
     /**
      * @return The name of the end point this request uses. Example: "campaign", "user", etc..
      */
     String getApiEndpoint();
 
+    /**
+     * @return The type of HTTP Request.
+     */
+    RequestMethod getRequestMethod();
+
     /**
      * @return correctly formatted request parameters.
      */
-    Map<String, String> getRequestParameters();
+    Object getRequestBody();
+
+    T parseResponse(final String responseStr) throws IOException;
 }

src/main/java/org/sourcelab/kafka/connect/apiclient/request/RequestErrorResponse.java

@@ -0,0 +1,22 @@
+package org.sourcelab.kafka.connect.apiclient.request;
+
+public class RequestErrorResponse {
+    private int errorCode;
+    private String message;
+
+    public int getErrorCode() {
+        return errorCode;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    @Override
+    public String toString() {
+        return "RequestErrorResponse{"
+            + "errorCode=" + errorCode
+            + ", message='" + message + '\''
+            + '}';
+    }
+}

src/main/java/org/sourcelab/kafka/connect/apiclient/request/RequestMethod.java

@@ -0,0 +1,7 @@
+package org.sourcelab.kafka.connect.apiclient.request;
+
+public enum RequestMethod {
+    GET,
+    POST,
+    PUT;
+}