1 files changed, 14 insertions, 10 deletions
diff --git a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
index d9efe802..02e153cf 100644
--- a/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
+++ b/datafile-app-server/src/main/java/org/onap/dcaegen2/collectors/datafile/tasks/DataRouterPublisher.java
@@ -22,19 +22,20 @@ package org.onap.dcaegen2.collectors.datafile.tasks;
 
 import com.google.gson.JsonElement;
 import com.google.gson.JsonParser;
+import java.io.File;
 import java.io.IOException;
-import java.io.InputStream;
 import java.net.MalformedURLException;
 import java.net.URI;
 import java.nio.file.Path;
 import java.time.Duration;
-import org.apache.commons.io.IOUtils;
 import org.apache.http.HttpResponse;
 import org.apache.http.client.methods.HttpPut;
-import org.apache.http.entity.ByteArrayEntity;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.FileEntity;
 import org.onap.dcaegen2.collectors.datafile.configuration.AppConfig;
 import org.onap.dcaegen2.collectors.datafile.configuration.PublisherConfiguration;
 import org.onap.dcaegen2.collectors.datafile.exceptions.DatafileTaskException;
+import org.onap.dcaegen2.collectors.datafile.model.Counters;
 import org.onap.dcaegen2.collectors.datafile.model.FilePublishInformation;
 import org.onap.dcaegen2.collectors.datafile.model.JsonSerializer;
 import org.onap.dcaegen2.collectors.datafile.model.logging.MappedDiagnosticContext;
@@ -62,9 +63,11 @@ public class DataRouterPublisher {
 
     private static final Logger logger = LoggerFactory.getLogger(DataRouterPublisher.class);
     private final AppConfig datafileAppConfig;
+    private final Counters counters;
 
-    public DataRouterPublisher(AppConfig datafileAppConfig) {
+    public DataRouterPublisher(AppConfig datafileAppConfig, Counters counters) {
         this.datafileAppConfig = datafileAppConfig;
+        this.counters = counters;
     }
 
     /**
@@ -98,8 +101,10 @@ public class DataRouterPublisher {
             HttpResponse response =
                     dmaapProducerHttpClient.getDmaapProducerResponseWithRedirect(put, publishInfo.getContext());
             logger.trace("{}", response);
+            counters.incTotalPublishedFiles();
             return Mono.just(HttpStatus.valueOf(response.getStatusLine().getStatusCode()));
         } catch (Exception e) {
+            counters.incNoOfFailedPublishAttempts();
             logger.warn("Publishing file {} to DR unsuccessful.", publishInfo.getName(), e);
             return Mono.error(e);
         }
@@ -121,10 +126,9 @@ public class DataRouterPublisher {
     }
 
     private void prepareBody(FilePublishInformation publishInfo, HttpPut put) throws IOException {
-        Path fileLocation = publishInfo.getInternalLocation();
-        try (InputStream fileInputStream = createInputStream(fileLocation)) {
-            put.setEntity(new ByteArrayEntity(IOUtils.toByteArray(fileInputStream)));
-        }
+        File file = createInputFile(publishInfo.getInternalLocation());
+        FileEntity entity = new FileEntity(file, ContentType.DEFAULT_BINARY);
+        put.setEntity(entity);
     }
 
     private static Mono<FilePublishInformation> handleHttpResponse(HttpStatus response,
@@ -140,9 +144,9 @@ public class DataRouterPublisher {
         }
     }
 
-    InputStream createInputStream(Path filePath) throws IOException {
+    File createInputFile(Path filePath) throws IOException {
         FileSystemResource realResource = new FileSystemResource(filePath);
-        return realResource.getInputStream();
+        return realResource.getFile();
     }
 
     PublisherConfiguration resolveConfiguration(String changeIdentifer) throws DatafileTaskException {