Publishing location operation response into iot location event stream

components/device-mgt/org.wso2.carbon.device.mgt.core/pom.xml

@@ -66,6 +66,7 @@
                             org.wso2.carbon.core,
                             org.wso2.carbon.utils.*,
                             org.wso2.carbon.device.mgt.common.*,
+                            org.wso2.carbon.device.mgt.analytics.data.publisher.service,
                             org.wso2.carbon.user.api,
                             org.wso2.carbon.user.core.*,
                             org.wso2.carbon.registry.core.service,
@@ -81,7 +82,7 @@
                             org.wso2.carbon.ntask.common,
                             org.apache.catalina,
                             org.apache.catalina.core,
-                            org.apache.commons.collections,
+                            org.apache.commons.collections;version="${commons-collections.version.range}",
                             org.wso2.carbon.email.sender.*,
                             io.swagger.annotations.*;resolution:=optional,
                             org.wso2.carbon,
@@ -136,6 +137,17 @@
             <groupId>org.wso2.carbon.devicemgt</groupId>
             <artifactId>org.wso2.carbon.device.mgt.common</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.wso2.carbon.devicemgt</groupId>
+            <artifactId>org.wso2.carbon.device.mgt.analytics.data.publisher</artifactId>
+            <scope>provided</scope>
+            <exclusions>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-api</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
         <dependency>
             <groupId>org.wso2.carbon</groupId>
             <artifactId>org.wso2.carbon.logging</artifactId>

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/device/details/mgt/impl/DeviceInformationManagerImpl.java

@@ -22,7 +22,11 @@ package org.wso2.carbon.device.mgt.core.device.details.mgt.impl;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.wso2.carbon.context.CarbonContext;
-import org.wso2.carbon.device.mgt.common.*;
+import org.wso2.carbon.device.mgt.analytics.data.publisher.exception.DataPublisherConfigurationException;
+import org.wso2.carbon.device.mgt.common.Device;
+import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
+import org.wso2.carbon.device.mgt.common.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.TransactionManagementException;
 import org.wso2.carbon.device.mgt.common.device.details.DeviceInfo;
 import org.wso2.carbon.device.mgt.common.device.details.DeviceLocation;
 import org.wso2.carbon.device.mgt.core.dao.DeviceDAO;
@@ -33,6 +37,7 @@ import org.wso2.carbon.device.mgt.core.device.details.mgt.DeviceInformationManag
 import org.wso2.carbon.device.mgt.core.device.details.mgt.dao.DeviceDetailsDAO;
 import org.wso2.carbon.device.mgt.core.device.details.mgt.dao.DeviceDetailsMgtDAOException;
 import org.wso2.carbon.device.mgt.core.internal.DeviceManagementDataHolder;
+import org.wso2.carbon.device.mgt.core.util.DeviceManagerUtil;
 
 import java.sql.SQLException;
 import java.util.ArrayList;
@@ -45,6 +50,7 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
     private DeviceDetailsDAO deviceDetailsDAO;
     private DeviceDAO deviceDAO;
     private static final Log log = LogFactory.getLog(DeviceInformationManagerImpl.class);
+    private static final String EVENT_STREAM_DEFINITION = "org.wso2.iot.LocationStream";
 
     public DeviceInformationManagerImpl() {
         this.deviceDAO = DeviceManagementDAOFactory.getDeviceDAO();
@@ -160,6 +166,17 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
             deviceDAO.updateDevice(device, CarbonContext.getThreadLocalCarbonContext().getTenantId());
             deviceDetailsDAO.deleteDeviceLocation(deviceLocation.getDeviceId());
             deviceDetailsDAO.addDeviceLocation(deviceLocation);
+            if (DeviceManagerUtil.isPublishLocationOperationResEnabled()) {
+                Object metaData[] = {device.getDeviceIdentifier(), device.getType()};
+                Object payload[] = new Object[]{
+                        deviceLocation.getUpdatedTime().getTime(),
+                        deviceLocation.getLatitude(),
+                        deviceLocation.getLongitude()
+                };
+                DeviceManagerUtil.getEventPublisherService().publishEvent(
+                        EVENT_STREAM_DEFINITION, "1.0.0", metaData, new Object[0], payload
+                );
+            }
             DeviceManagementDAOFactory.commitTransaction();
         } catch (TransactionManagementException e) {
             throw new DeviceDetailsMgtException("Transactional error occurred while adding the device location " +
@@ -174,6 +191,9 @@ public class DeviceInformationManagerImpl implements DeviceInformationManager {
             DeviceManagementDAOFactory.rollbackTransaction();
             throw new DeviceDetailsMgtException("Error occurred while updating the last updated timestamp of " +
                     "the device", e);
+        } catch (DataPublisherConfigurationException e) {
+            DeviceManagementDAOFactory.rollbackTransaction();
+            throw new DeviceDetailsMgtException("Error occurred while publishing the device location information.", e);
         } finally {
             DeviceManagementDAOFactory.closeConnection();
         }

components/device-mgt/org.wso2.carbon.device.mgt.core/src/main/java/org/wso2/carbon/device/mgt/core/util/DeviceManagerUtil.java

@@ -21,7 +21,15 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.w3c.dom.Document;
 import org.wso2.carbon.base.MultitenantConstants;
-import org.wso2.carbon.device.mgt.common.*;
+import org.wso2.carbon.context.PrivilegedCarbonContext;
+import org.wso2.carbon.device.mgt.analytics.data.publisher.service.EventsPublisherService;
+import org.wso2.carbon.device.mgt.common.Device;
+import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
+import org.wso2.carbon.device.mgt.common.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.EnrolmentInfo;
+import org.wso2.carbon.device.mgt.common.GroupPaginationRequest;
+import org.wso2.carbon.device.mgt.common.PaginationRequest;
+import org.wso2.carbon.device.mgt.common.TransactionManagementException;
 import org.wso2.carbon.device.mgt.common.group.mgt.GroupManagementException;
 import org.wso2.carbon.device.mgt.common.notification.mgt.NotificationManagementException;
 import org.wso2.carbon.device.mgt.common.operation.mgt.OperationManagementException;
@@ -48,7 +56,11 @@ import javax.xml.XMLConstants;
 import javax.xml.parsers.DocumentBuilder;
 import javax.xml.parsers.DocumentBuilderFactory;
 import java.io.File;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.Hashtable;
+import java.util.List;
+import java.util.Map;
 
 
 public final class DeviceManagerUtil {
@@ -358,6 +370,17 @@ public final class DeviceManagerUtil {
         return limit;
     }
 
+    public static boolean isPublishLocationOperationResEnabled() throws DeviceManagementException {
+        DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                getDeviceManagementConfig();
+        if (deviceManagementConfig != null) {
+            return deviceManagementConfig.getGeoLocationConfiguration().getPublishLocationOperationResponse();
+        } else {
+            throw new DeviceManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                        "cdm-config.xml file.");
+        }
+    }
+
     public static DeviceIDHolder validateDeviceIdentifiers(List<DeviceIdentifier> deviceIDs) {
 
         List<String> errorDeviceIdList = new ArrayList<String>();
@@ -404,4 +427,16 @@ public final class DeviceManagerUtil {
         }
         return true;
     }
+
+    public static EventsPublisherService getEventPublisherService() {
+        PrivilegedCarbonContext ctx = PrivilegedCarbonContext.getThreadLocalCarbonContext();
+        EventsPublisherService eventsPublisherService =
+                (EventsPublisherService) ctx.getOSGiService(EventsPublisherService.class, null);
+        if (eventsPublisherService == null) {
+            String msg = "Event Publisher service has not initialized.";
+            log.error(msg);
+            throw new IllegalStateException(msg);
+        }
+        return eventsPublisherService;
+    }
 }