Feature/get delete user data

example/src/main/java/EventsClientExample.java

@@ -29,8 +29,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
-import java.util.concurrent.*;
-import java.util.function.Function;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.stream.Collectors;
 
 /**
@@ -43,9 +43,25 @@ public class EventsClientExample {
 
     public static void main(String[] args) {
 
-        String serverHost = args[0]; // this should be https://... for TLS/SSL when Harness has TLS/SSL enabled
-        String engineId = args[1];
-        String fileName = args[2];
+        String serverHost = null;
+        try {
+            serverHost = args[0]; // this should be https://... for TLS/SSL when Harness has TLS/SSL enabled
+        } catch (Exception e) {
+            serverHost = "http://localhost";
+        }
+        String engineId = null;
+        try {
+            engineId = args[1];
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.exit(1);
+        }
+        String fileName = null;
+        try {
+            fileName = args[2];
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
         Integer serverPort = 9090;
 
         log.info("Args: {}, {}, {}, {}", engineId, fileName, serverHost, serverPort);
@@ -60,16 +76,21 @@ public static void main(String[] args) {
         Optional<Path> cert = Optional.empty();
         EventsClient client = new EventsClient(engineId, serverHost, serverPort, optionalCreds, cert);
 
+        runCreateEvent(client, engineId, fileName);
+    }
+
+
+    private static void runCreateEvent(EventsClient client, String engineId, String fileName) {
         // example of JSON for creating an event
         String json = "{" +
                 "\"eventId\":\"ed15537661f2492cab64615096c93160\"," +
                 "\"event\":\"$set\"," +
                 "\"entityType\":\"testGroup\"," +
                 "\"entityId\":\"9\"," +
                 "\"properties\":{" +
-                    "\"testPeriodStart\":\"2016-07-12T00:00:00.000+09:00\"," +
-                    "\"pageVariants\":[\"17\",\"18\"]," +
-                    "\"testPeriodEnd\":\"2016-08-31T00:00:00.000+09:00\"}," +
+                "\"testPeriodStart\":\"2016-07-12T00:00:00.000+09:00\"," +
+                "\"pageVariants\":[\"17\",\"18\"]," +
+                "\"testPeriodEnd\":\"2016-08-31T00:00:00.000+09:00\"}," +
                 "\"eventTime\":\"2016-07-12T16:08:49.677+09:00\"," +
                 "\"creationTime\":\"2016-07-12T07:09:58.273Z\"}";
 
@@ -163,7 +184,5 @@ public static void main(String[] args) {
             log.error("Oops, we have an unrecoverable error: ", e);
             client.close();
         }
-
     }
-
 }

example/src/main/java/HarnessClientExample.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright ActionML, LLC under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * ActionML licenses this file to You 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.
+ */
+
+import com.actionml.HarnessClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.PasswordAuthentication;
+import java.util.Map;
+import java.util.Optional;
+
+public class HarnessClientExample {
+    private static Logger log = LoggerFactory.getLogger(HarnessClientExample.class);
+
+    public static void main(String[] args) {
+
+        String serverHost = null;
+        try {
+            serverHost = args[0]; // this should be https://... for TLS/SSL when Harness has TLS/SSL enabled
+        } catch (Exception e) {
+            serverHost = "http://localhost";
+        }
+        String engineId = null;
+        try {
+            engineId = args[1];
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.exit(1);
+        }
+        String fileName = null;
+        try {
+            fileName = args[2];
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+        Integer serverPort = 9090;
+
+        log.info("Args: {}, {}, {}, {}", engineId, fileName, serverHost, serverPort);
+
+        Map<String, String> env = System.getenv();
+        Optional<String> optUsername = Optional.ofNullable(env.getOrDefault("HARNESS_CLIENT_USER_ID", null));
+        Optional<String> optPassword = Optional.ofNullable(env.getOrDefault("HARNESS_CLIENT_USER_SECRET", null));
+        Optional<PasswordAuthentication> optionalCreds = optUsername.flatMap(username -> optPassword.map(password ->
+                new PasswordAuthentication(username, password.toCharArray())
+        ));
+
+        HarnessClient client = new HarnessClient(engineId, serverHost, serverPort, optionalCreds);
+
+        runGetUserData(client, "u1");
+//        runDeleteUserData(client, "u1");
+    }
+
+
+    private static void runGetUserData(HarnessClient client, String userId) {
+        client.deleteEvents(userId).thenAccept(System.out::println);
+    }
+
+    private static void runDeleteUserData(HarnessClient client, String userId) {
+        client.deleteEvents(userId).thenAccept(System.out::println);
+    }
+}

src/main/java/com/actionml/EventsClient.java

@@ -26,6 +26,7 @@
 import com.actionml.entity.Event;
 
 import java.net.PasswordAuthentication;
+import java.net.URI;
 import java.nio.file.Path;
 import java.util.ArrayList;
 import java.util.List;

src/main/java/com/actionml/HarnessClient.java

@@ -17,14 +17,55 @@
 
 package com.actionml;
 
+import akka.http.javadsl.model.Uri;
+import akka.japi.Pair;
+import akka.stream.javadsl.Source;
+
+import java.net.PasswordAuthentication;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.CompletionStage;
+
 /**
  * @author The ActionML Team (<a href="http://actionml.com">http://actionml.com</a>)
  *         26.02.17 18:02
  */
-public class HarnessClient {
+public class HarnessClient extends RestClient {
+
+    public HarnessClient(String engineId, String host, Integer port, Optional<PasswordAuthentication> optionalCreds) {
+        super(host, port, Uri.create("/engines").addPathSegment(engineId).addPathSegment("entities"), optionalCreds, Optional.empty());
+    }
 
-    public HarnessClient(String host, Integer port, String clientId, String clientSecret) {
+    public CompletionStage<Pair<Integer, List<String>>> getEvents(String userId) {
+        return withAuth().thenCompose(optionalToken ->
+                Source.single(this.uri.addPathSegment(userId))
+                        .map(this::createGet)
+                        .zipWithIndex()
+                        .map(pair -> pair.copy(pair.first(), (Long) pair.second()))
+                        .via(this.poolClientFlow)
+                        .mapAsync(1, this::extractResponse)
+                        .mapAsync(1, this::extractResponses)
+                        .runFold(Pair.create(0, new ArrayList<>()), (acc, pair) -> {
+                            Integer code = pair.second().first();
+                            List<String> events = acc.second();
+                            events.add(pair.second().second());
+                            return Pair.create(code, events);
+                        }, this.materializer)
+        );
+    }
 
+    public CompletionStage<Integer> deleteEvents(String userId) {
+        return withAuth().thenCompose(optionalToken ->
+                Source.single(this.uri.addPathSegment(userId))
+                        .map(this::createDelete)
+                        .zipWithIndex()
+                        .map(pair -> pair.copy(pair.first(), (Long) pair.second()))
+                        .via(this.poolClientFlow)
+                        .mapAsync(1, this::extractResponse)
+                        .runFold(0, (a, i) -> i.second().status().intValue(), this.materializer)
+        );
     }
 
+
 }