Updated the broker version

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.wso2.carbon.devicemgt-plugins</groupId>
         <artifactId>mb-extensions</artifactId>
-        <version>3.0.25-SNAPSHOT</version>
+        <version>3.0.38-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/beans/BasePaginatedResult.java

@@ -0,0 +1,42 @@
+/*
+ *   Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ *   WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModelProperty;
+
+public class BasePaginatedResult {
+
+    /**
+     * Number of Resources returned.
+     */
+    @ApiModelProperty(
+            value = "Number of total resources.",
+            example = "1")
+    @JsonProperty("count")
+    private long count;
+
+    public long getCount() {
+        return count;
+    }
+
+    public void setCount(long count) {
+        this.count = count;
+    }
+}

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/beans/TopicList.java

@@ -0,0 +1,55 @@
+/*
+ *   Copyright (c) 2016, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ *   WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.annotations.ApiModelProperty;
+import org.wso2.carbon.andes.admin.mqtt.internal.xsd.Subscription;
+import org.wso2.carbon.device.mgt.common.Device;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class TopicList extends BasePaginatedResult {
+
+    private List<Subscription> subscriptions = new ArrayList<>();
+
+    @ApiModelProperty(value = "List of devices returned")
+    @JsonProperty("subscriptions")
+    public List<Subscription> getList() {
+        return subscriptions;
+    }
+
+    public void setList(List<Subscription> subscriptions) {
+        this.subscriptions = subscriptions;
+    }
+
+    @Override
+    public String toString() {
+        StringBuilder sb = new StringBuilder();
+        sb.append("{\n");
+
+        sb.append("  count: ").append(getCount()).append(",\n");
+        sb.append("  subscriptions: [").append(subscriptions).append("\n");
+        sb.append("]}\n");
+        return sb.toString();
+    }
+
+}
+

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/beans/TopicPaginationRequest.java

@@ -0,0 +1,114 @@
+/*
+ * Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans;
+
+import java.util.Date;
+
+/**
+ * This class holds required parameters for a querying a paginated device response.
+ */
+public class TopicPaginationRequest {
+
+    private int startIndex;
+    private int rowCount;
+    private Date since;
+    private String topic_name;
+    private int remaining_messages;
+    private String active;
+    private String durable;
+    private String subscriber_name;
+    private String identifier;
+
+    public TopicPaginationRequest(int start, int rowCount) {
+        this.startIndex = start;
+        this.rowCount = rowCount;
+    }
+
+    public int getStartIndex() {
+        return startIndex;
+    }
+
+    public void setStartIndex(int startIndex) {
+        this.startIndex = startIndex;
+    }
+
+    public int getRowCount() {
+        return rowCount;
+    }
+
+    public void setRowCount(int rowCount) {
+        this.rowCount = rowCount;
+    }
+
+    public Date getSince() {
+        return since;
+    }
+
+    public void setSince(Date since) {
+        this.since = since;
+    }
+
+    public String getTopic_name() {
+        return topic_name;
+    }
+
+    public void setTopic_name(String topic_name) {
+        this.topic_name = topic_name;
+    }
+
+    public int getRemaining_messages() {
+        return remaining_messages;
+    }
+
+    public void setRemaining_messages(int remaining_messages) {
+        this.remaining_messages = remaining_messages;
+    }
+
+    public String getActive() {
+        return active;
+    }
+
+    public void setActive(String active) {
+        this.active = active;
+    }
+
+    public String getDurable() {
+        return durable;
+    }
+
+    public void setDurable(String durable) {
+        this.durable = durable;
+    }
+
+    public String getSubscriber_name() {
+        return subscriber_name;
+    }
+
+    public void setSubscriber_name(String subscriber_name) {
+        this.subscriber_name = subscriber_name;
+    }
+
+    public String getIdentifier() {
+        return identifier;
+    }
+
+    public void setIdentifier(String identifier) {
+        this.identifier = identifier;
+    }
+}

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/exception/AuthenticationException.java

@@ -0,0 +1,74 @@
+/*
+ *   Copyright (c) 2015, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ *   WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.exception;
+
+public class AuthenticationException extends Exception {
+
+    private static final long serialVersionUID = -3151279311929070297L;
+
+    private String errorMessage;
+    private int errorCode;
+
+    public AuthenticationException(int errorCode, String message) {
+        super(message);
+        this.errorCode = errorCode;
+    }
+
+    public AuthenticationException(int errorCode, String message, Throwable cause) {
+        super(message, cause);
+        this.errorCode = errorCode;
+    }
+
+    public int getErrorCode() {
+        return errorCode;
+    }
+
+
+    public String getErrorMessage() {
+        return errorMessage;
+    }
+
+    public void setErrorMessage(String errorMessage) {
+        this.errorMessage = errorMessage;
+    }
+
+    public AuthenticationException(String msg, Exception nestedEx) {
+        super(msg, nestedEx);
+        setErrorMessage(msg);
+    }
+
+    public AuthenticationException(String message, Throwable cause) {
+        super(message, cause);
+        setErrorMessage(message);
+    }
+
+    public AuthenticationException(String msg) {
+        super(msg);
+        setErrorMessage(msg);
+    }
+
+    public AuthenticationException() {
+        super();
+    }
+
+    public AuthenticationException(Throwable cause) {
+        super(cause);
+    }
+    
+}

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/exception/TopicManagementException.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright (c) 2014, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.exception;
+
+public class TopicManagementException extends Exception {
+
+    private static final long serialVersionUID = -3151279311929070297L;
+
+    public TopicManagementException(String msg, Exception nestedEx) {
+        super(msg, nestedEx);
+    }
+
+    public TopicManagementException(String message, Throwable cause) {
+        super(message, cause);
+    }
+
+    public TopicManagementException(String msg) {
+        super(msg);
+    }
+
+    public TopicManagementException() {
+        super();
+    }
+
+    public TopicManagementException(Throwable cause) {
+        super(cause);
+    }
+
+}

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/service/api/admin/MQTTManagementAdminService.java

@@ -125,6 +125,12 @@ public interface MQTTManagementAdminService {
     })
     Response getFilteredSubscriptions(
 
+            @QueryParam("topic_name") String topic_name,
+            @QueryParam("remaining_messages") int remaining_messages,
+            @QueryParam("active") String active,
+            @QueryParam("durable") String durable,
+            @QueryParam("subscriber_name") String subscriber_name,
+            @QueryParam("identifier") String identifier,
             @ApiParam(
                     name = "tenant-domain",
                     value = "The name of the tenant.\n" +
@@ -132,6 +138,7 @@ public interface MQTTManagementAdminService {
                     required = true,
                     defaultValue = "carbon.super")
             @QueryParam("tenant-domain") String tenantDomain,
+            @QueryParam("since") String since,
             @ApiParam(
                     name = "If-Modified-Since",
                     value = "Checks if the requested variant was modified, since the specified date-time. \n" +

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/service/impl/admin/MQTTManagementAdminServiceImpl.java

@@ -19,25 +19,45 @@
 package org.wso2.carbon.andes.extensions.device.mgt.jaxrs.service.impl.admin;
 
 import org.apache.axis2.AxisFault;
+import org.apache.axis2.client.Stub;
 import org.apache.axis2.context.ConfigurationContext;
 import org.apache.axis2.transport.http.HTTPConstants;
 import org.apache.axis2.transport.http.HttpTransportProperties;
+import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.juddi.rmi.UDDISubscriptionService;
 import org.wso2.carbon.CarbonConstants;
 import org.wso2.carbon.andes.admin.mqtt.internal.xsd.Subscription;
 import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans.ErrorResponse;
+import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans.TopicList;
+import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans.TopicPaginationRequest;
+import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.exception.AuthenticationException;
 import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.service.api.admin.MQTTManagementAdminService;
 import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.service.impl.util.RequestValidationUtil;
+import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.service.impl.util.TopicManagerUtil;
 import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.util.MQTTMgtAPIUtils;
 import org.wso2.carbon.andes.mqtt.stub.AndesMQTTAdminServiceBrokerManagerAdminException;
 import org.wso2.carbon.andes.mqtt.stub.AndesMQTTAdminServiceStub;
-import org.wso2.carbon.base.MultitenantConstants;
 import org.wso2.carbon.context.CarbonContext;
 import org.wso2.carbon.context.PrivilegedCarbonContext;
-import org.wso2.carbon.device.mgt.common.DeviceManagementException;
+import org.wso2.carbon.device.mgt.common.*;
+import org.wso2.carbon.device.mgt.common.app.mgt.Application;
+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.DeviceManagementDAOException;
+import org.wso2.carbon.device.mgt.core.dao.DeviceManagementDAOFactory;
+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.user.api.UserStoreException;
+import org.wso2.carbon.user.core.service.RealmService;
 import org.wso2.carbon.utils.CarbonUtils;
+import org.wso2.carbon.utils.multitenancy.MultitenantConstants;
 
+
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.TrustManagerFactory;
 import javax.servlet.ServletConfig;
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpSession;
@@ -45,9 +65,15 @@ import javax.ws.rs.*;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.MediaType;
 import javax.ws.rs.core.Response;
+import java.io.File;
 import java.rmi.RemoteException;
-import java.util.HashMap;
-import java.util.Map;
+import java.security.*;
+import java.sql.SQLException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.*;
+
+import static sun.audio.AudioDevice.device;
 
 @Path("/admin/topics")
 @Produces(MediaType.APPLICATION_JSON)
@@ -55,6 +81,61 @@ import java.util.Map;
 public class MQTTManagementAdminServiceImpl implements MQTTManagementAdminService {
 
     private static final Log log = LogFactory.getLog(MQTTManagementAdminServiceImpl.class);
+    /**
+     * required soap header for authorization
+     */
+    private static final String AUTHORIZATION_HEADER = "Authorization";
+
+    /**
+     * required soap header value for mutualSSL
+     */
+    private static final String AUTHORIZATION_HEADER_VALUE = "Bearer";
+
+    private static final String KEY_STORE_TYPE = "JKS";
+    /**
+     * Default truststore type of the client
+     */
+    private static final String TRUST_STORE_TYPE = "JKS";
+    /**
+     * Default keymanager type of the client
+     */
+    private static final String KEY_MANAGER_TYPE = "SunX509"; //Default Key Manager Type
+    /**
+     * Default trustmanager type of the client
+     */
+    private static final String TRUST_MANAGER_TYPE = "SunX509"; //Default Trust Manager Type
+
+    private static final String SSLV3 = "SSLv3";
+
+
+
+    private KeyStore keyStore;
+    private KeyStore trustStore;
+    private char[] keyStorePassword;
+    private SSLContext sslContext;
+
+    private String tenantDomain;
+
+    private static final String DEFAULT_RESOURCE_LOCATION = "/resources/devicetypes";
+    private static final String CAR_FILE_LOCATION = CarbonUtils.getCarbonHome() + File.separator + "repository" +
+            File.separator + "resources" + File.separator + "devicetypes";
+    private static final String DAS_PORT = "${iot.analytics.https.port}";
+    private static final String DAS_HOST_NAME = "${iot.analytics.host}";
+    private static final String DEFAULT_HTTP_PROTOCOL = "https";
+    private static final String IOT_MGT_PORT = "${iot.manager.https.port}";
+    private static final String IOT_MGT_HOST_NAME = "${iot.manager.host}";
+    private static final String DAS_URL = DEFAULT_HTTP_PROTOCOL + "://" + DAS_HOST_NAME
+            + ":" + DAS_PORT + "/services/CarbonAppUploader/";
+    private static final String DAS_EVENT_RECEIVER_EP = DEFAULT_HTTP_PROTOCOL + "://" + DAS_HOST_NAME
+            + ":" + DAS_PORT + "/services/EventReceiverAdminService/";
+    private static final String DAS_EVENT_STREAM_EP = DEFAULT_HTTP_PROTOCOL + "://" + DAS_HOST_NAME
+            + ":" + DAS_PORT + "/services/EventStreamAdminService/";
+
+    private static final String IOT_MGT_URL = DEFAULT_HTTP_PROTOCOL + "://" + IOT_MGT_HOST_NAME
+            + ":" + IOT_MGT_PORT + "/services/CarbonAppUploader/";
+    private static final String MEDIA_TYPE_XML = "application/xml";
+    private static final String DEVICE_MANAGEMENT_TYPE = "device_management";
+    private static final String TENANT_DOMAIN_PROPERTY = "\\$\\{tenant-domain\\}";
     private
     @Context
     ServletConfig config;
@@ -65,25 +146,66 @@ public class MQTTManagementAdminServiceImpl implements MQTTManagementAdminServic
     @Override
     @GET
     public Response getFilteredSubscriptions(
+            @QueryParam("topic_name") String topic_name,
+            @QueryParam("remaining_messages") int remaining_messages,
+            @QueryParam("active") String active,
+            @QueryParam("durable") String durable,
+            @QueryParam("subscriber_name") String subscriber_name,
+            @QueryParam("identifier") String identifier,
             @QueryParam("tenant-domain") String tenantDomain,
+            @QueryParam("since") String since,
             @HeaderParam("If-Modified-Since") String ifModifiedSince,
             @QueryParam("offset") int offset,
             @QueryParam("limit") int limit) {
-        RequestValidationUtil.validatePaginationParameters(offset, limit);
         try {
+            if (!StringUtils.isEmpty(topic_name) && !StringUtils.isEmpty(active)) {
+                return Response.status(Response.Status.BAD_REQUEST).entity(
+                        new ErrorResponse.ErrorResponseBuilder().setMessage("Request contains both name and role " +
+                                "parameters. Only one is allowed " +
+                                "at once.").build()).build();
+            }
             int currentTenantId = CarbonContext.getThreadLocalCarbonContext().getTenantId();
             if (MultitenantConstants.SUPER_TENANT_ID != currentTenantId) {
                 return Response.status(Response.Status.UNAUTHORIZED).entity(
                         new ErrorResponse.ErrorResponseBuilder().setMessage(
                                 "Current logged in user is not authorized to perform this operation").build()).build();
             }
+            String ten_Domain = getTenantDomain(currentTenantId);
             PrivilegedCarbonContext.startTenantFlow();
-            PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(tenantDomain);
-            PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(MQTTMgtAPIUtils.getTenantId(tenantDomain));
+            PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(ten_Domain);
+            PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantId(MQTTMgtAPIUtils.getTenantId(ten_Domain));
 
+
+            RequestValidationUtil.validatePaginationParameters(offset, limit);
             HttpSession session = request.getSession();
-
             AndesMQTTAdminServiceStub andesAdminStub = getAndesMQTTAdminServiceStub(config, session, request);
+            TopicPaginationRequest request = new TopicPaginationRequest(offset, limit);
+            PaginationResult result;
+            TopicList topics = new TopicList();
+
+            if (topic_name != null && !topic_name.isEmpty()) {
+                request.setTopic_name(topic_name);
+            }
+            if (remaining_messages != 0) {
+                request.setRemaining_messages(remaining_messages);
+            }
+            if (active != null && !active.isEmpty()) {
+                request.setActive(active);
+            }
+            if (durable != null && !durable.isEmpty()) {
+                request.setDurable(durable);
+            }
+            if (subscriber_name != null && !subscriber_name.isEmpty()) {
+                request.setSubscriber_name(subscriber_name);
+            }
+            if (identifier != null && !identifier.isEmpty()) {
+                request.setIdentifier(identifier);
+            }
+
+//            Subscription[] filteredNormalTopicSubscriptionList = andesAdminStub.getFilteredMQTTSubscriptions(false, true,
+//                    "MQTT", "TOPIC", "", false,
+//                    "", false, "All", offset,
+//                    10,currentTenantId,tenantDomain);
 
             Subscription[] filteredNormalTopicSubscriptionList = andesAdminStub.getFilteredSubscriptions(false, true,
                     "MQTT", "TOPIC", "", false,
@@ -91,6 +213,55 @@ public class MQTTManagementAdminServiceImpl implements MQTTManagementAdminServic
                     10);
             Map<String, Subscription[]> subscriptions = new HashMap<>();
             subscriptions.put("subscriptions", filteredNormalTopicSubscriptionList);
+
+            //getPaginatedTopics(request,subscriptions);
+
+            if (ifModifiedSince != null && !ifModifiedSince.isEmpty()) {
+                Date sinceDate;
+                SimpleDateFormat format = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss Z");
+                try {
+                    sinceDate = format.parse(ifModifiedSince);
+                } catch (ParseException e) {
+                    return Response.status(Response.Status.BAD_REQUEST).entity(
+                            new ErrorResponse.ErrorResponseBuilder().setMessage("Invalid date " +
+                                    "string is provided in 'If-Modified-Since' header").build()).build();
+                }
+                request.setSince(sinceDate);
+                result = getPaginatedTopics(request,subscriptions);
+                if (result == null || result.getData() == null || result.getData().size() <= 0) {
+                    return Response.status(Response.Status.NOT_MODIFIED).entity("No device is modified " +
+                            "after the timestamp provided in 'If-Modified-Since' header").build();
+                }
+            } else if (since != null && !since.isEmpty()) {
+                Date sinceDate;
+                SimpleDateFormat format = new SimpleDateFormat("EEE, d MMM yyyy HH:mm:ss Z");
+                try {
+                    sinceDate = format.parse(since);
+                } catch (ParseException e) {
+                    return Response.status(Response.Status.BAD_REQUEST).entity(
+                            new ErrorResponse.ErrorResponseBuilder().setMessage("Invalid date " +
+                                    "string is provided in 'since' filter").build()).build();
+                }
+                request.setSince(sinceDate);
+                result = getPaginatedTopics(request,subscriptions);
+                if (result == null || result.getData() == null || result.getData().size() <= 0) {
+                    topics.setList(new ArrayList<Subscription>());
+                    topics.setCount(0);
+                    return Response.status(Response.Status.OK).entity(topics).build();
+                }
+            } else {
+                result = getPaginatedTopics(request,subscriptions);
+                int resultCount = result.getRecordsTotal();
+                if (resultCount == 0) {
+                    Response.status(Response.Status.OK).entity(topics).build();
+                }
+            }
+
+            topics.setList((List<Subscription>) result.getData());
+            topics.setCount(result.getRecordsTotal());
+
+
+
             return Response.ok().entity(subscriptions).build();
         } catch (DeviceManagementException e) {
             String msg = "Error occurred at server side while fetching device list.";
@@ -112,6 +283,11 @@ public class MQTTManagementAdminServiceImpl implements MQTTManagementAdminServic
             log.error(msg, e);
             return Response.serverError().entity(
                     new ErrorResponse.ErrorResponseBuilder().setMessage(msg).build()).build();
+        } catch (AuthenticationException e) {
+            String msg = "Error occurred at server side while fetching service stub.";
+            log.error(msg, e);
+            return Response.serverError().entity(
+                    new ErrorResponse.ErrorResponseBuilder().setMessage(msg).build()).build();
         } finally {
             PrivilegedCarbonContext.endTenantFlow();
         }
@@ -150,4 +326,79 @@ public class MQTTManagementAdminServiceImpl implements MQTTManagementAdminServic
         return stub;
     }
 
+
+    private PaginationResult getPaginatedTopics(TopicPaginationRequest request,Map<String, Subscription[]> subscriptions) throws DeviceManagementException {
+        List<Subscription> devicesForRoles = null;
+        PaginationResult paginationResult = new PaginationResult();
+        List<Subscription> allSubscriptions =  new ArrayList<>();
+        List<Subscription> sub =  new ArrayList<>();
+        Map<String,Subscription[]> ss = subscriptions;
+        int count = 0;
+        int tenantId = this.getTenantId();
+        request = TopicManagerUtil.validateTopicListPageSize(request);
+
+        List<Subscription[]> subscriptionList = new ArrayList<Subscription[]>(subscriptions.values());
+
+        allSubscriptions = Arrays.asList(subscriptionList.get(0));
+        count = allSubscriptions.size();
+
+        paginationResult.setData(allSubscriptions);
+        paginationResult.setRecordsFiltered(count);
+        paginationResult.setRecordsTotal(count);
+        return paginationResult;
+    }
+
+    private int getTenantId() {
+        return CarbonContext.getThreadLocalCarbonContext().getTenantId();
+    }
+
+    private static String getTenantDomain(int tenantId) throws AuthenticationException {
+        try {
+            PrivilegedCarbonContext.startTenantFlow();
+            PrivilegedCarbonContext ctx = PrivilegedCarbonContext.getThreadLocalCarbonContext();
+
+            RealmService realmService = (RealmService) ctx.getOSGiService(RealmService.class, null);
+            if (realmService == null) {
+                String msg = "RealmService is not initialized";
+                log.error(msg);
+                throw new AuthenticationException(msg);
+            }
+
+            return realmService.getTenantManager().getDomain(tenantId);
+
+        } catch (UserStoreException e) {
+            String msg = "User store not initialized";
+            log.error(msg);
+            throw new AuthenticationException(msg, e);
+        } finally {
+            PrivilegedCarbonContext.endTenantFlow();
+        }
+    }
+
+    /**
+     * Initializes the SSL Context
+     */
+    private void initSSLConnection() throws NoSuchAlgorithmException, UnrecoverableKeyException,
+            KeyStoreException, KeyManagementException {
+        KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KEY_MANAGER_TYPE);
+        keyManagerFactory.init(keyStore, keyStorePassword);
+        TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance(TRUST_MANAGER_TYPE);
+        trustManagerFactory.init(trustStore);
+
+        // Create and initialize SSLContext for HTTPS communication
+        sslContext = SSLContext.getInstance(SSLV3);
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+        SSLContext.setDefault(sslContext);
+    }
+
+    private void cleanup(Stub stub) {
+        if (stub != null) {
+            try {
+                stub.cleanup();
+            } catch (AxisFault axisFault) {
+                //do nothing
+            }
+        }
+    }
+
 }

components/extensions/mb-extensions/org.wso2.carbon.andes.extensions.device.mgt.api/src/main/java/org/wso2/carbon/andes/extensions/device/mgt/jaxrs/service/impl/util/TopicManagerUtil.java

@@ -0,0 +1,409 @@
+/*
+ * Copyright (c) 2014, WSO2 Inc. (http://www.wso2.org) All Rights Reserved.
+ *
+ * WSO2 Inc. 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 org.wso2.carbon.andes.extensions.device.mgt.jaxrs.service.impl.util;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.w3c.dom.Document;
+import org.wso2.carbon.andes.extensions.device.mgt.jaxrs.beans.TopicPaginationRequest;
+import org.wso2.carbon.base.MultitenantConstants;
+import org.wso2.carbon.device.mgt.common.*;
+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;
+import org.wso2.carbon.device.mgt.core.config.DeviceConfigurationManager;
+import org.wso2.carbon.device.mgt.core.config.DeviceManagementConfig;
+import org.wso2.carbon.device.mgt.core.config.datasource.DataSourceConfig;
+import org.wso2.carbon.device.mgt.core.config.datasource.JNDILookupDefinition;
+import org.wso2.carbon.device.mgt.core.dao.DeviceManagementDAOException;
+import org.wso2.carbon.device.mgt.core.dao.DeviceManagementDAOFactory;
+import org.wso2.carbon.device.mgt.core.dao.DeviceTypeDAO;
+import org.wso2.carbon.device.mgt.core.dao.util.DeviceManagementDAOUtil;
+import org.wso2.carbon.device.mgt.core.dto.DeviceType;
+import org.wso2.carbon.device.mgt.core.internal.DeviceManagementDataHolder;
+import org.wso2.carbon.device.mgt.core.operation.mgt.OperationMgtConstants;
+import org.wso2.carbon.device.mgt.core.operation.mgt.util.DeviceIDHolder;
+import org.wso2.carbon.device.mgt.core.util.DeviceManagerUtil;
+import org.wso2.carbon.user.api.TenantManager;
+import org.wso2.carbon.user.api.UserStoreException;
+import org.wso2.carbon.utils.CarbonUtils;
+import org.wso2.carbon.utils.ConfigurationContextService;
+import org.wso2.carbon.utils.NetworkUtils;
+
+import javax.sql.DataSource;
+import javax.xml.XMLConstants;
+import javax.xml.parsers.DocumentBuilder;
+import javax.xml.parsers.DocumentBuilderFactory;
+import java.io.File;
+import java.util.*;
+
+
+public final class TopicManagerUtil {
+
+    private static final Log log = LogFactory.getLog(DeviceManagerUtil.class);
+
+    public static Document convertToDocument(File file) throws DeviceManagementException {
+        DocumentBuilderFactory factory = DocumentBuilderFactory.newInstance();
+        factory.setNamespaceAware(true);
+        try {
+            DocumentBuilder docBuilder = factory.newDocumentBuilder();
+            factory.setFeature(XMLConstants.FEATURE_SECURE_PROCESSING, true);
+            return docBuilder.parse(file);
+        } catch (Exception e) {
+            throw new DeviceManagementException("Error occurred while parsing file, while converting " +
+                    "to a org.w3c.dom.Document", e);
+        }
+    }
+
+    /**
+     * Resolve data source from the data source definition.
+     *
+     * @param config data source configuration
+     * @return data source resolved from the data source definition
+     */
+    public static DataSource resolveDataSource(DataSourceConfig config) {
+        DataSource dataSource = null;
+        if (config == null) {
+            throw new RuntimeException("Device Management Repository data source configuration is null and thus, " +
+                    "is not initialized");
+        }
+        JNDILookupDefinition jndiConfig = config.getJndiLookupDefinition();
+        if (jndiConfig != null) {
+            if (log.isDebugEnabled()) {
+                log.debug("Initializing Device Management Repository data source using the JNDI Lookup Definition");
+            }
+            List<JNDILookupDefinition.JNDIProperty> jndiPropertyList =
+                    jndiConfig.getJndiProperties();
+            if (jndiPropertyList != null) {
+                Hashtable<Object, Object> jndiProperties = new Hashtable<Object, Object>();
+                for (JNDILookupDefinition.JNDIProperty prop : jndiPropertyList) {
+                    jndiProperties.put(prop.getName(), prop.getValue());
+                }
+                dataSource = DeviceManagementDAOUtil.lookupDataSource(jndiConfig.getJndiName(), jndiProperties);
+            } else {
+                dataSource = DeviceManagementDAOUtil.lookupDataSource(jndiConfig.getJndiName(), null);
+            }
+        }
+        return dataSource;
+    }
+
+    /**
+     * Adds a new device type to the database if it does not exists.
+     *
+     * @param typeName device type
+     * @param tenantId provider tenant Id
+     * @param isSharedWithAllTenants is this device type shared with all tenants.
+     * @return status of the operation
+     */
+    public static boolean registerDeviceType(String typeName, int tenantId, boolean isSharedWithAllTenants)
+            throws DeviceManagementException {
+        boolean status;
+        try {
+            DeviceManagementDAOFactory.beginTransaction();
+            DeviceTypeDAO deviceTypeDAO = DeviceManagementDAOFactory.getDeviceTypeDAO();
+            DeviceType deviceType = deviceTypeDAO.getDeviceType(typeName, tenantId);
+            if (deviceType == null) {
+                deviceType = new DeviceType();
+                deviceType.setName(typeName);
+                deviceTypeDAO.addDeviceType(deviceType, tenantId, isSharedWithAllTenants);
+            }
+            DeviceManagementDAOFactory.commitTransaction();
+            status = true;
+        } catch (DeviceManagementDAOException e) {
+            DeviceManagementDAOFactory.rollbackTransaction();
+            throw new DeviceManagementException("Error occurred while registering the device type '"
+                                                        + typeName + "'", e);
+        } catch (TransactionManagementException e) {
+            DeviceManagementDAOFactory.rollbackTransaction();
+            throw new DeviceManagementException("SQL occurred while registering the device type '"
+                                                        + typeName + "'", e);
+        } finally {
+            DeviceManagementDAOFactory.closeConnection();
+        }
+        return status;
+    }
+
+    /**
+     * Un-registers an existing device type from the device management metadata repository.
+     *
+     * @param typeName device type
+     * @return status of the operation
+     */
+    public static boolean unregisterDeviceType(String typeName, int tenantId) throws DeviceManagementException {
+        try {
+            DeviceManagementDAOFactory.beginTransaction();
+            DeviceTypeDAO deviceTypeDAO = DeviceManagementDAOFactory.getDeviceTypeDAO();
+            DeviceType deviceType = deviceTypeDAO.getDeviceType(typeName, tenantId);
+            if (deviceType != null) {
+                deviceTypeDAO.removeDeviceType(typeName, tenantId);
+            }
+            DeviceManagementDAOFactory.commitTransaction();
+            return true;
+        } catch (DeviceManagementDAOException e) {
+            DeviceManagementDAOFactory.rollbackTransaction();
+            throw new DeviceManagementException("Error occurred while registering the device type '" +
+                                                        typeName + "'", e);
+        } catch (TransactionManagementException e) {
+            DeviceManagementDAOFactory.rollbackTransaction();
+            throw new DeviceManagementException("SQL occurred while registering the device type '" +
+                                                        typeName + "'", e);
+        } finally {
+            DeviceManagementDAOFactory.closeConnection();
+        }
+    }
+
+    public static Map<String, String> convertDevicePropertiesToMap(List<Device.Property> properties) {
+        Map<String, String> propertiesMap = new HashMap<String, String>();
+        for (Device.Property prop : properties) {
+            propertiesMap.put(prop.getName(), prop.getValue());
+        }
+        return propertiesMap;
+    }
+
+    public static List<DeviceIdentifier> convertDevices(List<Device> devices) {
+
+        List<DeviceIdentifier> deviceIdentifiers = new ArrayList<>();
+        for (Device device : devices) {
+            DeviceIdentifier identifier = new DeviceIdentifier();
+            identifier.setId(device.getDeviceIdentifier());
+            identifier.setType(device.getType());
+            deviceIdentifiers.add(identifier);
+        }
+        return deviceIdentifiers;
+    }
+
+    public static List<DeviceIdentifier> getValidDeviceIdentifiers(List<Device> devices) {
+        List<DeviceIdentifier> deviceIdentifiers = new ArrayList<>();
+        for (Device device : devices) {
+            if (device.getEnrolmentInfo() != null) {
+                switch (device.getEnrolmentInfo().getStatus()) {
+                    case BLOCKED:
+                    case REMOVED:
+                    case SUSPENDED:
+                        break;
+                    default:
+                        DeviceIdentifier identifier = new DeviceIdentifier();
+                        identifier.setId(device.getDeviceIdentifier());
+                        identifier.setType(device.getType());
+                        deviceIdentifiers.add(identifier);
+                }
+            }
+        }
+        return deviceIdentifiers;
+    }
+
+
+    public static String getServerBaseHttpsUrl() {
+        String hostName = "localhost";
+        try {
+            hostName = NetworkUtils.getMgtHostName();
+        } catch (Exception ignored) {
+        }
+        String mgtConsoleTransport = CarbonUtils.getManagementTransport();
+        ConfigurationContextService configContextService =
+                DeviceManagementDataHolder.getInstance().getConfigurationContextService();
+        int port = CarbonUtils.getTransportPort(configContextService, mgtConsoleTransport);
+        int httpsProxyPort =
+                CarbonUtils.getTransportProxyPort(configContextService.getServerConfigContext(),
+                        mgtConsoleTransport);
+        if (httpsProxyPort > 0) {
+            port = httpsProxyPort;
+        }
+        return "https://" + hostName + ":" + port;
+    }
+
+    public static String getServerBaseHttpUrl() {
+        String hostName = "localhost";
+        try {
+            hostName = NetworkUtils.getMgtHostName();
+        } catch (Exception ignored) {
+        }
+        ConfigurationContextService configContextService =
+                DeviceManagementDataHolder.getInstance().getConfigurationContextService();
+        int port = CarbonUtils.getTransportPort(configContextService, "http");
+        int httpProxyPort =
+                CarbonUtils.getTransportProxyPort(configContextService.getServerConfigContext(),
+                        "http");
+        if (httpProxyPort > 0) {
+            port = httpProxyPort;
+        }
+        return "http://" + hostName + ":" + port;
+    }
+
+    /**
+     * returns the tenant Id of the specific tenant Domain
+     *
+     * @param tenantDomain
+     * @return
+     * @throws DeviceManagementException
+     */
+    public static int getTenantId(String tenantDomain) throws DeviceManagementException {
+        try {
+            if (MultitenantConstants.SUPER_TENANT_DOMAIN_NAME.equals(tenantDomain)) {
+                return MultitenantConstants.SUPER_TENANT_ID;
+            }
+            TenantManager tenantManager = DeviceManagementDataHolder.getInstance().getTenantManager();
+            int tenantId = tenantManager.getTenantId(tenantDomain);
+            if (tenantId == -1) {
+                throw new DeviceManagementException("invalid tenant Domain :" + tenantDomain);
+            }
+            return tenantId;
+        } catch (UserStoreException e) {
+            throw new DeviceManagementException("invalid tenant Domain :" + tenantDomain);
+        }
+    }
+
+    public static int validateActivityListPageSize(int limit) throws OperationManagementException {
+        if (limit == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                    getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                return deviceManagementConfig.getPaginationConfiguration().getActivityListPageSize();
+            } else {
+                throw new OperationManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                    "cdm-config.xml file.");
+            }
+        }
+        return limit;
+    }
+
+    public static PaginationRequest validateOperationListPageSize(PaginationRequest paginationRequest) throws
+                                                                                          OperationManagementException {
+        if (paginationRequest.getRowCount() == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                    getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                paginationRequest.setRowCount(deviceManagementConfig.getPaginationConfiguration().
+                        getOperationListPageSize());
+            } else {
+                throw new OperationManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                    "cdm-config.xml file.");
+            }
+        }
+        return paginationRequest;
+    }
+
+    public static PaginationRequest validateNotificationListPageSize(PaginationRequest paginationRequest) throws
+                                                                                       NotificationManagementException {
+        if (paginationRequest.getRowCount() == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                    getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                paginationRequest.setRowCount(deviceManagementConfig.getPaginationConfiguration().
+                        getNotificationListPageSize());
+            } else {
+                throw new NotificationManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                          "cdm-config.xml file.");
+            }
+        }
+        return paginationRequest;
+    }
+
+        public static TopicPaginationRequest validateTopicListPageSize(TopicPaginationRequest paginationRequest) throws
+                                                                                             DeviceManagementException {
+        if (paginationRequest.getRowCount() == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                    getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                paginationRequest.setRowCount(deviceManagementConfig.getPaginationConfiguration().
+                        getTopicListPageSize());
+            } else {
+                throw new DeviceManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                    "cdm-config.xml file.");
+            }
+        }
+        return paginationRequest;
+    }
+
+    public static GroupPaginationRequest validateGroupListPageSize(GroupPaginationRequest paginationRequest) throws
+                                                                                                    GroupManagementException {
+        if (paginationRequest.getRowCount() == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance()
+                    .getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                paginationRequest.setRowCount(deviceManagementConfig.getPaginationConfiguration()
+                                                      .getDeviceListPageSize());
+            } else {
+                throw new GroupManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                   "cdm-config.xml file.");
+            }
+        }
+        return paginationRequest;
+    }
+
+    public static int validateDeviceListPageSize(int limit) throws DeviceManagementException {
+        if (limit == 0) {
+            DeviceManagementConfig deviceManagementConfig = DeviceConfigurationManager.getInstance().
+                    getDeviceManagementConfig();
+            if (deviceManagementConfig != null) {
+                return deviceManagementConfig.getPaginationConfiguration().getDeviceListPageSize();
+            } else {
+                throw new DeviceManagementException("Device-Mgt configuration has not initialized. Please check the " +
+                                                    "cdm-config.xml file.");
+            }
+        }
+        return limit;
+    }
+
+    public static DeviceIDHolder validateDeviceIdentifiers(List<DeviceIdentifier> deviceIDs) {
+
+        List<String> errorDeviceIdList = new ArrayList<String>();
+        List<DeviceIdentifier> validDeviceIDList = new ArrayList<DeviceIdentifier>();
+
+        int deviceIDCounter = 0;
+        for (DeviceIdentifier deviceIdentifier : deviceIDs) {
+
+            deviceIDCounter++;
+            String deviceID = deviceIdentifier.getId();
+
+            if (deviceID == null || deviceID.isEmpty()) {
+                errorDeviceIdList.add(String.format(OperationMgtConstants.DeviceConstants.DEVICE_ID_NOT_FOUND,
+                        deviceIDCounter));
+                continue;
+            }
+
+            try {
+
+                if (isValidDeviceIdentifier(deviceIdentifier)) {
+                    validDeviceIDList.add(deviceIdentifier);
+                } else {
+                    errorDeviceIdList.add(deviceID);
+                }
+            } catch (DeviceManagementException e) {
+                errorDeviceIdList.add(deviceID);
+            }
+        }
+
+        DeviceIDHolder deviceIDHolder = new DeviceIDHolder();
+        deviceIDHolder.setValidDeviceIDList(validDeviceIDList);
+        deviceIDHolder.setErrorDeviceIdList(errorDeviceIdList);
+
+        return deviceIDHolder;
+    }
+
+    public static boolean isValidDeviceIdentifier(DeviceIdentifier deviceIdentifier) throws DeviceManagementException {
+        Device device = DeviceManagementDataHolder.getInstance().getDeviceManagementProvider().getDevice(deviceIdentifier);
+        if (device == null || device.getDeviceIdentifier() == null ||
+                device.getDeviceIdentifier().isEmpty() || device.getEnrolmentInfo() == null) {
+            return false;
+        } else if (EnrolmentInfo.Status.REMOVED.equals(device.getEnrolmentInfo().getStatus())) {
+            return false;
+        }
+        return true;
+    }
+}

features/extensions-feature/org.wso2.carbon.andes.extensions.device.mgt.api.feature/pom.xml

@@ -24,7 +24,7 @@
     <parent>
         <groupId>org.wso2.carbon.devicemgt-plugins</groupId>
         <artifactId>extensions-feature</artifactId>
-        <version>3.0.25-SNAPSHOT</version>
+        <version>3.0.38-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

pom.xml

@@ -1040,7 +1040,7 @@
             <dependency>
                 <groupId>org.wso2.andes.wso2</groupId>
                 <artifactId>andes</artifactId>
-                <version>${carbon.messaging.version}</version>
+                <version>${andes.version}</version>
             </dependency>
             <dependency>
                 <groupId>org.wso2.carbon.devicemgt-plugins</groupId>
@@ -1263,9 +1263,10 @@
         <javax.xml.parsers.import.pkg.version>[0.0.0,1.0.0)</javax.xml.parsers.import.pkg.version>
 
         <!-- MB Features -->
-        <carbon.messaging.version>3.1.3</carbon.messaging.version>
+        <carbon.messaging.version>3.2.6-SNAPSHOT</carbon.messaging.version>
+        <andes.version>3.2.7-SNAPSHOT</andes.version>
         <jackson-annotations.version>2.6.1.wso2v1</jackson-annotations.version>
-        <carbon.andes.mqtt.version>3.2.1-SNAPSHOT</carbon.andes.mqtt.version>
+        <carbon.andes.mqtt.version>3.2.6-SNAPSHOT</carbon.andes.mqtt.version>
 
         <!--Feign Version-->
         <io.github.openfeign.version>9.3.1</io.github.openfeign.version>