Add improvements for the reporting data publishing

Co-authored-by: Pahansith Gunathilake <pahansith@entgra.io>
Co-committed-by: Pahansith Gunathilake <pahansith@entgra.io>

components/device-mgt/io.entgra.device.mgt.core.device.mgt.core/src/main/java/io/entgra/device/mgt/core/device/mgt/core/app/mgt/ApplicationManagerProviderServiceImpl.java

@@ -19,6 +19,7 @@
 package io.entgra.device.mgt.core.device.mgt.core.app.mgt;
 
 import com.google.gson.Gson;
+import io.entgra.device.mgt.core.device.mgt.core.report.mgt.ReportingPublisherManager;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -286,8 +287,9 @@ public class ApplicationManagerProviderServiceImpl implements ApplicationManagem
                 deviceDetailsWrapper.setTenantId(tenantId);
                 deviceDetailsWrapper.setDevice(device);
                 deviceDetailsWrapper.setApplications(newApplications);
-                HttpReportingUtil.invokeApi(deviceDetailsWrapper.getJSONString(),
+                ReportingPublisherManager reportingManager = new ReportingPublisherManager();
-                                            reportingHost + DeviceManagementConstants.Report.APP_USAGE_ENDPOINT);
+                reportingManager.publishData(deviceDetailsWrapper, DeviceManagementConstants
+                        .Report.APP_USAGE_ENDPOINT);
             }
 
         } catch (DeviceManagementDAOException e) {

components/device-mgt/io.entgra.device.mgt.core.device.mgt.core/src/main/java/io/entgra/device/mgt/core/device/mgt/core/device/details/mgt/impl/DeviceInformationManagerImpl.java

@@ -18,6 +18,7 @@
 
 package io.entgra.device.mgt.core.device.mgt.core.device.details.mgt.impl;
 
+import io.entgra.device.mgt.core.device.mgt.core.report.mgt.ReportingPublisherManager;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -29,7 +30,6 @@ import io.entgra.device.mgt.core.device.mgt.common.device.details.DeviceDetailsW
 import io.entgra.device.mgt.core.device.mgt.common.device.details.DeviceInfo;
 import io.entgra.device.mgt.core.device.mgt.common.device.details.DeviceLocation;
 import io.entgra.device.mgt.core.device.mgt.common.exceptions.DeviceManagementException;
-import io.entgra.device.mgt.core.device.mgt.common.exceptions.EventPublishingException;
 import io.entgra.device.mgt.core.device.mgt.common.exceptions.TransactionManagementException;
 import io.entgra.device.mgt.core.device.mgt.common.group.mgt.DeviceGroup;
 import io.entgra.device.mgt.core.device.mgt.common.group.mgt.GroupManagementException;
@@ -54,6 +54,8 @@ import java.util.Calendar;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 
 public class DeviceInformationManagerImpl implements DeviceInformationManager {
 
@@ -86,6 +88,7 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
 
             DeviceDetailsWrapper deviceDetailsWrapper = new DeviceDetailsWrapper();
             deviceDetailsWrapper.setDeviceInfo(deviceInfo);
+            //Asynchronous call to publish the device information to the reporting service. Hence, response is ignored.
             publishEvents(device, deviceDetailsWrapper, DeviceManagementConstants.Report.DEVICE_INFO_PARAM);
 
             DeviceManagementDAOFactory.beginTransaction();
@@ -203,12 +206,34 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
                     getDeviceManagementProvider().getDevice(deviceIdentifier, false);
             DeviceDetailsWrapper deviceDetailsWrapper = new DeviceDetailsWrapper();
             deviceDetailsWrapper.setEvents(payload);
-            return publishEvents(device, deviceDetailsWrapper, eventType);
+            Future<Integer> apiCallback = publishEvents(device, deviceDetailsWrapper, eventType);
+            if (null != apiCallback) {
+                boolean isDebugEnabled = log.isDebugEnabled();
+                while(!apiCallback.isDone()) {
+                    if (isDebugEnabled) {
+                        log.debug("Waiting for the response from the API for the reporting data " +
+                                "publishing for the device " + deviceId + ". Event payload: " + payload);
+                    }
+                }
+                return apiCallback.get();
+            }
+            return 0; // If the event publishing is disabled.
         } catch (DeviceManagementException e) {
             DeviceManagementDAOFactory.rollbackTransaction();
             String msg = "Event publishing error. Could not get device " + deviceId;
             log.error(msg, e);
             throw new DeviceDetailsMgtException(msg, e);
+        } catch (ExecutionException e) {
+            //Exceptions thrown in ReportingPublisherManager will be wrapped under this exception
+            String message = "Failed while publishing device information data to the reporting service for the device "
+                    + deviceId;
+            log.error(message, e);
+            throw new DeviceDetailsMgtException(message, e);
+        } catch (InterruptedException e) {
+            String message = "Failed while publishing device information data to the reporting service. Thread " +
+                    "interrupted while waiting for the response from the API for the Device " + deviceId;
+            log.error(message, e);
+            throw new DeviceDetailsMgtException(message, e);
         }
     }
 
@@ -217,7 +242,7 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
      * @param device Device that is sending event
      * @param deviceDetailsWrapper Payload to send(example, deviceinfo, applist, raw events)
      */
-    private int publishEvents(Device device, DeviceDetailsWrapper deviceDetailsWrapper, String
+    private Future<Integer> publishEvents(Device device, DeviceDetailsWrapper deviceDetailsWrapper, String
             eventType)  {
         String reportingHost = HttpReportingUtil.getReportingHost();
         if (!StringUtils.isBlank(reportingHost)
@@ -252,9 +277,8 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
 
                 String eventUrl = reportingHost + DeviceManagementConstants.Report
                         .REPORTING_CONTEXT + DeviceManagementConstants.URL_SEPERATOR + eventType;
-                return HttpReportingUtil.invokeApi(deviceDetailsWrapper.getJSONString(), eventUrl);
+                ReportingPublisherManager reportingManager = new ReportingPublisherManager();
-            } catch (EventPublishingException e) {
+                return reportingManager.publishData(deviceDetailsWrapper, eventUrl);
-                log.error("Error occurred while sending events", e);
             } catch (GroupManagementException e) {
                 log.error("Error occurred while getting group list", e);
             } catch (UserStoreException e) {
@@ -270,7 +294,7 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
                         + DeviceManagerUtil.getTenantId());
             }
         }
-        return 0;
+        return null;
     }
 
     @Override

components/device-mgt/io.entgra.device.mgt.core.device.mgt.core/src/main/java/io/entgra/device/mgt/core/device/mgt/core/report/mgt/ReportingPublisherManager.java

@@ -0,0 +1,87 @@
+/*
+ * Copyright (c) 2018 - 2024, Entgra (Pvt) Ltd. (http://www.entgra.io) All Rights Reserved.
+ *
+ * Entgra (Pvt) Ltd. 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.
+ */
+
+package io.entgra.device.mgt.core.device.mgt.core.report.mgt;
+
+import io.entgra.device.mgt.core.device.mgt.common.device.details.DeviceDetailsWrapper;
+import io.entgra.device.mgt.core.device.mgt.common.exceptions.EventPublishingException;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.http.HttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.entity.ContentType;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
+import org.apache.http.protocol.HTTP;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+public class ReportingPublisherManager {
+
+    private static final Log log = LogFactory.getLog(ReportingPublisherManager.class);
+    private final static ExecutorService executorService;
+    private DeviceDetailsWrapper payload;
+    private String endpoint;
+    private static final PoolingHttpClientConnectionManager poolingManager;
+
+    static {
+        executorService = Executors.newFixedThreadPool(10); //todo make this configurable
+        poolingManager = new PoolingHttpClientConnectionManager();
+        poolingManager.setMaxTotal(10); //todo make this configurable
+        poolingManager.setDefaultMaxPerRoute(10);
+    }
+
+    public Future<Integer> publishData(DeviceDetailsWrapper deviceDetailsWrapper, String eventUrl) {
+        this.payload = deviceDetailsWrapper;
+        this.endpoint = eventUrl;
+        return executorService.submit(new ReportingPublisher());
+    }
+
+    private class ReportingPublisher implements Callable<Integer> {
+        @Override
+        public Integer call() throws EventPublishingException {
+            try (CloseableHttpClient client = HttpClients.custom().setConnectionManager(poolingManager).build()) {
+                HttpPost apiEndpoint = new HttpPost(endpoint);
+                apiEndpoint.setHeader(HTTP.CONTENT_TYPE, ContentType.APPLICATION_JSON.toString());
+                StringEntity requestEntity = new StringEntity(payload.getJSONString(), ContentType.APPLICATION_JSON);
+                apiEndpoint.setEntity(requestEntity);
+                HttpResponse response = client.execute(apiEndpoint);
+                int statusCode = response.getStatusLine().getStatusCode();
+                if (log.isDebugEnabled()) {
+                    log.debug("Published data to the reporting backend: " + endpoint + ", Response code: " + statusCode);
+                }
+                return statusCode;
+            } catch (ConnectException e) {
+                String message = "Connection refused while publishing reporting data to the API: " + endpoint;
+                log.error(message, e);
+                throw new EventPublishingException(message, e);
+            } catch (IOException e) {
+                String message = "Error occurred when publishing reporting data to the API: " + endpoint;
+                log.error(message, e);
+                throw new EventPublishingException(message, e);
+            }
+        }
+    }
+}

components/device-mgt/io.entgra.device.mgt.core.device.mgt.core/src/main/java/io/entgra/device/mgt/core/device/mgt/core/util/HttpReportingUtil.java

@@ -27,6 +27,7 @@ import org.apache.http.entity.ContentType;
 import org.apache.http.entity.StringEntity;
 import org.apache.http.impl.client.CloseableHttpClient;
 import org.apache.http.impl.client.HttpClients;
+import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
 import org.apache.http.protocol.HTTP;
 import org.json.JSONObject;
 import io.entgra.device.mgt.core.device.mgt.common.exceptions.EventPublishingException;
@@ -49,24 +50,6 @@ public class HttpReportingUtil {
         return System.getProperty(DeviceManagementConstants.Report.REPORTING_EVENT_HOST);
     }
 
-    public static int invokeApi(String payload, String endpoint) throws EventPublishingException {
-        try (CloseableHttpClient client = HttpClients.createDefault()) {
-            HttpPost apiEndpoint = new HttpPost(endpoint);
-            apiEndpoint.setHeader(HTTP.CONTENT_TYPE, ContentType.APPLICATION_JSON.toString());
-            StringEntity requestEntity = new StringEntity(
-                    payload, ContentType.APPLICATION_JSON);
-            apiEndpoint.setEntity(requestEntity);
-            HttpResponse response = client.execute(apiEndpoint);
-            return response.getStatusLine().getStatusCode();
-        } catch (ConnectException e) {
-            log.error("Connection refused to API endpoint: " + endpoint, e);
-            return HttpStatus.SC_SERVICE_UNAVAILABLE;
-        } catch (IOException e) {
-            throw new EventPublishingException("Error occurred when " +
-                    "invoking API. API endpoint: " + endpoint, e);
-        }
-    }
-
 
     public static boolean isPublishingEnabledForTenant() {
         Object configuration = DeviceManagerUtil.getConfiguration(IS_EVENT_PUBLISHING_ENABLED);