Restapi-call-node: Fix sending big files to DMAAP data router

Issue-ID: CCSDK-2122
Signed-off-by: Stan Bonev <sb5356@att.com>
Change-Id: If71e6d3ee0bd649994ce4b9a23dcbd1b42a64101

3 files changed, 112 insertions, 63 deletions

diff --git a/restapi-call-node/provider/src/main/java/org/onap/ccsdk/sli/plugins/restapicall/RestapiCallNode.java b/restapi-call-node/provider/src/main/java/org/onap/ccsdk/sli/plugins/restapicall/RestapiCallNode.java
index 165eefde4..709774bb9 100755
--- a/restapi-call-node/provider/src/main/java/org/onap/ccsdk/sli/plugins/restapicall/RestapiCallNode.java
+++ b/restapi-call-node/provider/src/main/java/org/onap/ccsdk/sli/plugins/restapicall/RestapiCallNode.java
@@ -25,15 +25,22 @@ package org.onap.ccsdk.sli.plugins.restapicall;
 import static java.lang.Boolean.valueOf;
 import static javax.ws.rs.client.Entity.entity;
 import static org.onap.ccsdk.sli.plugins.restapicall.AuthType.fromString;
+import java.io.BufferedReader;
 import java.io.File;
 import java.io.FileInputStream;
 import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStream;
+import java.net.HttpURLConnection;
+import java.net.ProtocolException;
 import java.net.SocketException;
 import java.net.URI;
+import java.net.URL;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.security.KeyStore;
 import java.util.ArrayList;
+import java.util.Base64;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -72,7 +79,9 @@ import org.glassfish.jersey.media.multipart.file.FileDataBodyPart;
 import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
 import org.onap.ccsdk.sli.core.sli.SvcLogicException;
 import org.onap.ccsdk.sli.core.sli.SvcLogicJavaPlugin;
+import org.onap.logging.filter.base.HttpURLConnectionMetricUtil;
 import org.onap.logging.filter.base.MetricLogClientFilter;
+import org.onap.logging.filter.base.ONAPComponents;
 import org.onap.logging.ref.slf4j.ONAPLogConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -128,6 +137,7 @@ public class RestapiCallNode implements SvcLogicJavaPlugin {
         httpReadTimeout = readOptionalInteger("HTTP_READ_TIMEOUT_MS",DEFAULT_HTTP_READ_TIMEOUT_MS);
     }
 
+    @SuppressWarnings("unchecked")
     protected void loadPartners(JSONObject partners) {
         Iterator<String> keys = partners.keys();
         String partnerUserKey = "user";
@@ -956,6 +966,16 @@ public class RestapiCallNode implements SvcLogicJavaPlugin {
             byte[] data = Files.readAllBytes(Paths.get(p.fileName));
 
             r = sendHttpData(data, p);
+
+            for (int i = 0; i < 10 && r.code == 301; i++) {
+                String newUrl = r.headers2.get("Location").get(0);
+
+                log.info("Got response code 301. Sending same request to URL: " + newUrl);
+
+                p.url = newUrl;
+                r = sendHttpData(data, p);
+            }
+
             setResponseStatus(ctx, p.responsePrefix, r);
 
         } catch (SvcLogicException | IOException e) {
@@ -1030,12 +1050,27 @@ public class RestapiCallNode implements SvcLogicJavaPlugin {
         }
     }
 
-    protected HttpResponse sendHttpData(byte[] data, FileParam p) throws SvcLogicException {
+    protected HttpResponse sendHttpData(byte[] data, FileParam p) throws IOException {
+        URL url = new URL(p.url);
+        HttpURLConnection con = (HttpURLConnection) url.openConnection();
 
-        Client client = ClientBuilder.newBuilder().build();
-        setClientTimeouts(client);
-        client.property(ClientProperties.FOLLOW_REDIRECTS, true);
-        WebTarget webTarget = addAuthType(client, p).target(p.url);
+        log.info("Connection: " + con.getClass().getName());
+
+        con.setRequestMethod(p.httpMethod.toString());
+        con.setRequestProperty("Content-Type", "application/octet-stream");
+        con.setRequestProperty("Accept", "*/*");
+        con.setRequestProperty("Expect", "100-continue");
+        con.setFixedLengthStreamingMode(data.length);
+        con.setInstanceFollowRedirects(false);
+
+        if (p.user != null && p.password != null) {
+            String authString = p.user + ":" + p.password;
+            String authStringEnc = Base64.getEncoder().encodeToString(authString.getBytes());
+            con.setRequestProperty("Authorization", "Basic " + authStringEnc);
+        }
+
+        con.setDoInput(true);
+        con.setDoOutput(true);
 
         log.info("Sending file");
         long t1 = System.currentTimeMillis();
@@ -1044,69 +1079,46 @@ public class RestapiCallNode implements SvcLogicJavaPlugin {
         r.code = 200;
 
         if (!p.skipSending) {
-            String tt = "application/octet-stream";
-            Invocation.Builder invocationBuilder = webTarget.request(tt).accept(tt);
+            HttpURLConnectionMetricUtil util = new HttpURLConnectionMetricUtil();
+            util.logBefore(con, ONAPComponents.DMAAP);
 
-            Response response;
+            con.connect();
 
+            boolean continue100failed = false;
             try {
-                if (p.httpMethod == HttpMethod.POST) {
-                    response = invocationBuilder.post(Entity.entity(data, tt));
-                } else if (p.httpMethod == HttpMethod.PUT) {
-                    response = invocationBuilder.put(Entity.entity(data, tt));
-                } else {
-                    throw new SvcLogicException("Http operation" + p.httpMethod + "not supported");
-                }
-            } catch (ProcessingException e) {
-                throw new SvcLogicException(requestPostingException + e.getLocalizedMessage(), e);
-            }
-
-            r.code = response.getStatus();
-            r.headers = response.getStringHeaders();
-            EntityTag etag = response.getEntityTag();
-            if (etag != null) {
-                r.message = etag.getValue();
-            }
-            if (response.hasEntity() && r.code != 204) {
-                r.body = response.readEntity(String.class);
+                OutputStream os = con.getOutputStream();
+                os.write(data);
+                os.flush();
+                os.close();
+            } catch (ProtocolException e) {
+                continue100failed = true;
             }
 
-            if (r.code == 301) {
-                String newUrl = response.getStringHeaders().getFirst("Location");
+            r.code = con.getResponseCode();
+            r.headers2 = con.getHeaderFields();
 
-                log.info("Got response code 301. Sending same request to URL: {}", newUrl);
-
-                webTarget = client.target(newUrl);
-                invocationBuilder = webTarget.request(tt).accept(tt);
-
-                try {
-                    if (p.httpMethod == HttpMethod.POST) {
-                        response = invocationBuilder.post(Entity.entity(data, tt));
-                    } else if (p.httpMethod == HttpMethod.PUT) {
-                        response = invocationBuilder.put(Entity.entity(data, tt));
-                    } else {
-                        throw new SvcLogicException("Http operation" + p.httpMethod + "not supported");
-                    }
-                } catch (ProcessingException e) {
-                    throw new SvcLogicException(requestPostingException + e.getLocalizedMessage(), e);
+            if (r.code != 204 && !continue100failed) {
+                BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream()));
+                String inputLine;
+                StringBuffer response = new StringBuffer();
+                while ((inputLine = in.readLine()) != null) {
+                    response.append(inputLine);
                 }
+                in.close();
 
-                r.code = response.getStatus();
-                etag = response.getEntityTag();
-                if (etag != null) {
-                    r.message = etag.getValue();
-                }
-                if (response.hasEntity() && r.code != 204) {
-                    r.body = response.readEntity(String.class);
-                }
+                r.body = response.toString();
             }
+
+            util.logAfter(con);
+
+            con.disconnect();
         }
 
         long t2 = System.currentTimeMillis();
-        log.info(responseReceivedMessage, t2 - t1);
-        log.info(responseHttpCodeMessage, r.code);
+        log.info("Response received. Time: {}", t2 - t1);
+        log.info("HTTP response code: {}", r.code);
         log.info("HTTP response message: {}", r.message);
-        logHeaders(r.headers);
+        logHeaders(r.headers2);
         log.info("HTTP response: {}", r.body);
 
         return r;
@@ -1154,6 +1166,26 @@ public class RestapiCallNode implements SvcLogicJavaPlugin {
         }
     }
 
+    private void logHeaders(Map<String, List<String>> mm) {
+        if (mm == null || mm.isEmpty()) {
+            return;
+        }
+
+        List<String> ll = new ArrayList<>();
+        for (String s : mm.keySet()) {
+            if (s != null) {
+                ll.add(s);
+            }
+        }
+        Collections.sort(ll);
+
+        for (String name : ll) {
+            List<String> v = mm.get(name);
+            log.info("--- {}:{}", name, String.valueOf(mm.get(name)));
+            log.info("--- " + name + ": " + (v.size() == 1 ? v.get(0) : v));
+        }
+    }
+
     protected HttpResponse postOnUeb(String request, UebParam p) throws SvcLogicException {
         String[] urls = uebServers.split(" ");
         for (int i = 0; i < urls.length; i++) {
diff --git a/restapi-call-node/provider/src/test/java/org/onap/ccsdk/sli/plugins/restapicall/TestRestapiCallNode.java b/restapi-call-node/provider/src/test/java/org/onap/ccsdk/sli/plugins/restapicall/TestRestapiCallNode.java
index da7b80ee1..b2f75bcd1 100755
--- a/restapi-call-node/provider/src/test/java/org/onap/ccsdk/sli/plugins/restapicall/TestRestapiCallNode.java
+++ b/restapi-call-node/provider/src/test/java/org/onap/ccsdk/sli/plugins/restapicall/TestRestapiCallNode.java
@@ -24,12 +24,9 @@ package org.onap.ccsdk.sli.plugins.restapicall;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
-
 import java.util.HashMap;
 import java.util.Map;
-
 import org.codehaus.jettison.json.JSONObject;
-
 import org.junit.Before;
 import org.junit.Test;
 import org.onap.ccsdk.sli.core.sli.SvcLogicContext;
@@ -81,6 +78,21 @@ public class TestRestapiCallNode {
     }
 
     @Test
+    public void testSendFile() throws SvcLogicException {
+        SvcLogicContext ctx = new SvcLogicContext();
+
+        Map<String, String> p = new HashMap<>();
+        p.put("fileName", "src/test/resources/test_file.txt");
+        p.put("url", "https://testurl.test");
+        p.put("user", "user");
+        p.put("password", "*******");
+        p.put("skipSending", "true"); // Set real url, user, password, when testing actual sending
+
+        RestapiCallNode rcn = new RestapiCallNode();
+        rcn.sendFile(p, ctx);
+    }
+
+    @Test
     public void testJsonTemplate() throws SvcLogicException {
         SvcLogicContext ctx = new SvcLogicContext();
         ctx.setAttribute("tmp.sdn-circuit-req-row_length", "3");
@@ -479,16 +491,16 @@ public class TestRestapiCallNode {
         assertNull(rcn.partnerStore.get("partnerThree"));
 
         //In this scenario the caller expects username, password and url to be picked up from the partners json
-        Map<String, String> paramMap = new HashMap<String,String>();
+        Map<String, String> paramMap = new HashMap<>();
         paramMap.put("partner", partnerTwoKey);
 	rcn.handlePartner(paramMap );
-        assertEquals(partnerTwoUsername,paramMap.get(rcn.restapiUserKey));
-        assertEquals(partnerTwoPassword,paramMap.get(rcn.restapiPasswordKey));
-        assertEquals("http://localhost:7002",paramMap.get(rcn.restapiUrlString));
+        assertEquals(partnerTwoUsername,paramMap.get(RestapiCallNode.restapiUserKey));
+        assertEquals(partnerTwoPassword,paramMap.get(RestapiCallNode.restapiPasswordKey));
+        assertEquals("http://localhost:7002",paramMap.get(RestapiCallNode.restapiUrlString));
 
         //In this scenario the caller expects username, password and url to be picked up from the partners json
         //the provided suffix will be appended to the default url from the partners json
-        paramMap = new HashMap<String,String>();
+        paramMap = new HashMap<>();
         paramMap.put("partner", partnerTwoKey);
         paramMap.put("restapiUrlSuffix", "/networking/v1/instance/3");
 	rcn.handlePartner(paramMap);
diff --git a/restapi-call-node/provider/src/test/resources/test_file.txt b/restapi-call-node/provider/src/test/resources/test_file.txt
new file mode 100644
index 000000000..038d757dc
--- /dev/null
+++ b/restapi-call-node/provider/src/test/resources/test_file.txt
@@ -0,0 +1,5 @@
+“You think your pain and your heartbreak are unprecedented in the history of the world, but then you read. It was books that taught me that the things that tormented me most were the very things that connected me with all the people who were alive, who had ever been alive.” – James Baldwin
+“When I have a little money, I buy books; and if I have any left, I buy food and clothes.” – Erasmus
+“If you only read the books that everyone else is reading, you can only think what everyone else is thinking.” – Haruki Murakami
+“You don’t have to burn books to destroy a culture. Just get people to stop reading them.” – Ray Bradbury
+“A reader lives a thousand lives before he dies, said Jojen. The man who never reads lives only one.” – George R.R. Martin