Restapi caller

data-layers/data-layer-api-caller/pom.xml

@@ -0,0 +1,65 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <artifactId>aesop</artifactId>
+        <groupId>com.flipkart.aesop</groupId>
+        <version>1.2.1-SNAPSHOT</version>
+        <relativePath>../../pom.xml</relativePath>
+    </parent>
+    <artifactId>data-layer-api-caller</artifactId>
+    <dependencies>
+        <dependency>
+            <groupId>org.springframework</groupId>
+            <artifactId>spring-jdbc</artifactId>
+            <version>${org.springframework.version}</version>
+        </dependency>
+        <!-- aesop dependencies -->
+        <dependency>
+            <groupId>com.flipkart.aesop</groupId>
+            <artifactId>runtime-client-cluster</artifactId>
+            <version>1.2.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.flipkart.aesop</groupId>
+            <artifactId>client-event-consumer</artifactId>
+            <version>1.2.1-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.google.code.gson</groupId>
+            <artifactId>gson</artifactId>
+            <version>2.2.2</version>
+        </dependency>
+        <dependency>
+            <groupId>flipkart.scm</groupId>
+            <artifactId>java-cas-client</artifactId>
+            <version>0.0.17.4-auth</version>
+        </dependency>
+        <dependency>
+            <groupId>com.ning</groupId>
+            <artifactId>async-http-client</artifactId>
+            <version>1.8.15</version>
+        </dependency>
+        <dependency>
+            <groupId>com.flipkart.kloud.authn</groupId>
+            <artifactId>client</artifactId>
+            <version>15.07.17.1</version>
+        </dependency>
+        <dependency>
+            <groupId>com.flipkart.kloud.authn</groupId>
+            <artifactId>dropwizard8-relying-party</artifactId>
+            <version>15.07.17.1</version>
+        </dependency>
+        <dependency>
+            <groupId>javax.ws.rs</groupId>
+            <artifactId>javax.ws.rs-api</artifactId>
+            <version>2.0</version>
+        </dependency>
+        <dependency>
+            <groupId>flipkart.platform</groupId>
+            <artifactId>prometheus</artifactId>
+            <version>2.3.2</version>
+        </dependency>
+    </dependencies>
+
+
+</project>

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/ApiCallerDataLayer.java

@@ -0,0 +1,73 @@
+package com.flipkart.aesop.apicallerdatalayer;
+
+import com.flipkart.aesop.apicallerdatalayer.client.HttpPostClient;
+import com.flipkart.aesop.apicallerdatalayer.headers.ContextualHeaderProvider;
+import com.flipkart.aesop.apicallerdatalayer.headers.StaticHeaderProvider;
+import com.flipkart.aesop.event.AbstractEvent;
+import com.flipkart.aesop.processor.DestinationEventProcessor;
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import com.linkedin.databus.client.pub.ConsumerCallbackResult;
+import org.trpr.platform.core.impl.logging.LogFactory;
+import org.trpr.platform.core.spi.logging.Logger;
+
+import javax.naming.OperationNotSupportedException;
+import javax.ws.rs.core.Response;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 08/12/15.
+ */
+public class ApiCallerDataLayer implements DestinationEventProcessor{
+    private String url;
+    private StaticHeaderProvider staticHeaderProvider;
+    private ContextualHeaderProvider contextualHeaderProvider;
+    private HttpPostClient httpPostClient;
+    private static final Logger LOGGER = LogFactory.getLogger(ApiCallerDataLayer.class);
+
+    public void setStaticHeaderProvider(StaticHeaderProvider staticHeaderProvider) {
+        this.staticHeaderProvider = staticHeaderProvider;
+    }
+
+    public void setContextualHeaderProvider(ContextualHeaderProvider contextualHeaderProvider) {
+        this.contextualHeaderProvider = contextualHeaderProvider;
+    }
+
+    public void setHttpPostClient(HttpPostClient httpPostClient) {
+        this.httpPostClient = httpPostClient;
+    }
+
+    public void setUrl(String url) {
+        this.url = url;
+    }
+
+
+    @Override
+    public ConsumerCallbackResult processDestinationEvent(AbstractEvent destinationEvent) throws OperationNotSupportedException {
+        try {
+            Map<String, Object> eventMap = destinationEvent.getFieldMapPair();
+            Gson gson = new GsonBuilder().serializeNulls().create();
+            String payload = gson.toJson(eventMap);
+            Map<String, String> headers = new HashMap<String, String>();
+            headers.putAll(staticHeaderProvider.getHeaders());
+            headers.putAll(contextualHeaderProvider.getHeaders(eventMap));
+            headers.put("Content-Type", "application/json");
+            LOGGER.info("Making a post call to url: " + url + " with payload as: " + payload + " and headers as: " + headers);
+            Response response = httpPostClient.post(url, payload, headers);
+            int responseCode = response.getStatus();
+            if (responseCode >= 200 && responseCode < 300) {
+                LOGGER.info("Call successful with response code as " + responseCode + " for payload: " + payload);
+                return ConsumerCallbackResult.SUCCESS;
+            } else {
+                LOGGER.info("Call unsuccessful with response code as " + responseCode + " and message as " + response.readEntity(String.class) + " for payload: " + payload);
+                return ConsumerCallbackResult.ERROR;
+            }
+        }catch(Exception e){
+            LOGGER.error("Call unsuccessful with error: ",e);
+            return ConsumerCallbackResult.ERROR;
+        }
+    }
+
+
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/client/HttpPostClient.java

@@ -0,0 +1,15 @@
+package com.flipkart.aesop.apicallerdatalayer.client;
+
+
+import javax.ws.rs.core.Response;
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 09/12/15.
+ */
+public interface HttpPostClient {
+    //Request request = new Request(url,payload,headers);
+    //Response response = client.executePost(request);
+
+    public Response post(String url, String payload, Map<String,String> headers);
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/client/implementation/HttpCASPostClientImpl.java

@@ -0,0 +1,37 @@
+package com.flipkart.aesop.apicallerdatalayer.client.implementation;
+
+import com.flipkart.aesop.apicallerdatalayer.client.HttpPostClient;
+import com.flipkart.casclient.client.HttpAuthClient;
+import com.flipkart.casclient.entity.Request;
+import flipkart.platform.cachefarm.Cache;
+import org.trpr.platform.core.impl.logging.LogFactory;
+import org.trpr.platform.core.spi.logging.Logger;
+
+import javax.ws.rs.core.Response;
+import java.io.IOException;
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 09/12/15.
+ */
+public class HttpCASPostClientImpl implements HttpPostClient {
+    private HttpAuthClient httpAuthClient;
+    private static final Logger LOGGER = LogFactory.getLogger(HttpCASPostClientImpl.class);
+    public HttpCASPostClientImpl(String casUrl, String user, String password, boolean enableAuth, Cache cache) {
+        this.httpAuthClient = new HttpAuthClient(casUrl,user,password,enableAuth,cache);
+    }
+
+    @Override
+    public Response post(String url, String payload, Map<String, String> headers) {
+        Request request = new Request(url,payload,headers);
+        com.ning.http.client.Response response = httpAuthClient.executePost(request);
+        String responseBody;
+        try {
+            responseBody = response.getResponseBody();
+        } catch (IOException e) {
+            LOGGER.error("Unable to get response body for payload"+payload+" Keeping it blank.",e);
+            responseBody = "";
+        }
+        return Response.status(response.getStatusCode()).entity(responseBody).build();
+    }
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/client/implementation/HttpJavaxPostClientImpl.java

@@ -0,0 +1,28 @@
+package com.flipkart.aesop.apicallerdatalayer.client.implementation;
+
+
+import com.flipkart.aesop.apicallerdatalayer.client.HttpPostClient;
+import org.trpr.platform.core.impl.logging.LogFactory;
+import org.trpr.platform.core.spi.logging.Logger;
+
+import javax.ws.rs.client.Client;
+import javax.ws.rs.client.ClientBuilder;
+import javax.ws.rs.client.Entity;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 10/12/15.
+ */
+public class HttpJavaxPostClientImpl implements HttpPostClient {
+    Client client = ClientBuilder.newClient();
+    private static final Logger LOGGER = LogFactory.getLogger(HttpJavaxPostClientImpl.class);
+    @Override
+    public Response post(String url, String payload, Map<String, String> headers) {
+
+        Response response = client.target(url).request().post(Entity.entity(payload, MediaType.APPLICATION_JSON_TYPE));
+        LOGGER.info(response.readEntity(String.class));
+        return response;
+    }
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/headers/ContextualHeaderProvider.java

@@ -0,0 +1,10 @@
+package com.flipkart.aesop.apicallerdatalayer.headers;
+
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 14/12/15.
+ */
+public interface ContextualHeaderProvider {
+    public Map<String, String> getHeaders(Map<String, Object> event);
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/headers/StaticHeaderProvider.java

@@ -0,0 +1,10 @@
+package com.flipkart.aesop.apicallerdatalayer.headers;
+
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 14/12/15.
+ */
+public interface StaticHeaderProvider {
+    public Map<String, String> getHeaders();
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/headers/impl/StringBasedContextualHeaderProviderImpl.java

@@ -0,0 +1,27 @@
+package com.flipkart.aesop.apicallerdatalayer.headers.impl;
+
+import com.flipkart.aesop.apicallerdatalayer.headers.ContextualHeaderProvider;
+import com.google.common.base.Splitter;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 14/12/15.
+ */
+public class StringBasedContextualHeaderProviderImpl implements ContextualHeaderProvider {
+    Map<String,String> headers;
+
+    public StringBasedContextualHeaderProviderImpl(String headers) {
+        this.headers = Splitter.on(",").withKeyValueSeparator("=").split(headers);
+    }
+
+    @Override
+    public Map<String, String> getHeaders(Map<String, Object> event) {
+        Map<String,String> map = new HashMap<String,String>();
+        for(Map.Entry<String,String> entry : headers.entrySet()){
+            map.put(entry.getKey(),(String)event.get(entry.getValue()));
+        }
+        return map;
+    }
+}

data-layers/data-layer-api-caller/src/main/java/com/flipkart/aesop/apicallerdatalayer/headers/impl/StringBasedStaticHeaderProviderImpl.java

@@ -0,0 +1,22 @@
+package com.flipkart.aesop.apicallerdatalayer.headers.impl;
+
+import com.flipkart.aesop.apicallerdatalayer.headers.StaticHeaderProvider;
+import com.google.common.base.Splitter;
+
+import java.util.Map;
+
+/**
+ * Created by aman.gupta on 14/12/15.
+ */
+public class StringBasedStaticHeaderProviderImpl implements StaticHeaderProvider {
+    Map<String,String> headers;
+
+    public StringBasedStaticHeaderProviderImpl(String headers) {
+        this.headers = Splitter.on(",").withKeyValueSeparator("=").split(headers);
+    }
+
+    @Override
+    public Map<String, String> getHeaders() {
+        return headers;
+    }
+}

pom.xml

@@ -90,6 +90,7 @@
         <module>data-layers/data-layer-console-appender</module>
         <module>data-layers/data-layer-elastic-search</module>
         <module>data-layers/data-layer-kafka</module>
+        <module>data-layers/data-layer-api-caller</module>
         <module>samples/sample-memory-relay</module>
         <module>samples/sample-hbase-relay</module>
         <module>samples/sample-mysql-mysql-blocking-bootstrap</module>
@@ -100,6 +101,7 @@
         <module>samples/sample-snapshot-serializer</module>
         <module>samples/sample-mysql-relay</module>
         <module>samples/sample-client-common</module>
+        <module>samples/sample-api-caller-client-cluster-consumer</module>
         <module>samples/sample-console-appender-client-cluster-consumer</module>
         <module>samples/sample-elastic-search-client-cluster-consumer</module>
         <module>samples/sample-hbase-client-cluster-consumer</module>