Streaming URI fix for 2.0.5, bug #1201

Author: aebadirad

marklogic-data-hub/src/main/java/com/marklogic/hub/collector/impl/CollectorImpl.java

@@ -22,6 +22,8 @@
 import org.springframework.http.HttpMethod;
 import org.springframework.http.MediaType;
 import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
+import org.springframework.web.client.RequestCallback;
+import org.springframework.web.client.ResponseExtractor;
 import org.springframework.web.client.RestTemplate;
 
 import javax.naming.InvalidNameException;
@@ -40,10 +42,7 @@
 import java.security.cert.Certificate;
 import java.security.cert.CertificateParsingException;
 import java.security.cert.X509Certificate;
-import java.util.ArrayList;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 public class CollectorImpl implements Collector {
     private DatabaseClient client = null;
@@ -118,18 +117,22 @@ public DiskQueue<String> run(String jobId, String entity, String flow, int threa
                 uriString += "&options=" + URLEncoder.encode(objectMapper.writeValueAsString(options), "UTF-8");
             }
             URI uri = new URI(uriString);
-            HttpHeaders headers = new HttpHeaders();
-            headers.set("Accept", MediaType.TEXT_PLAIN_VALUE);
-            Resource responseBody = template.exchange(uri, HttpMethod.GET, new HttpEntity<>(headers), Resource.class).getBody();
-            if(responseBody != null) {
-                InputStream inputStream = responseBody.getInputStream();
-                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(inputStream));
+            RequestCallback requestCallback = request -> request.getHeaders()
+                .setAccept(Arrays.asList(MediaType.APPLICATION_OCTET_STREAM, MediaType.ALL));
+
+            // Streams the response instead of loading it all in memory
+            ResponseExtractor<Void> responseExtractor = response -> {
+                BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(response.getBody()));
                 String line;
                 while((line = bufferedReader.readLine()) != null) {
                     results.add(line);
                 }
-                inputStream.close();
-            }
+                bufferedReader.close();
+                return null;
+            };
+
+            template.execute(uri, HttpMethod.GET, requestCallback, responseExtractor);
+
             return results;
         }
         catch(Exception e) {