forked from community/product-iots
GPrathap
9 years ago
parent
7ab6046d31
commit
8a013279a4
@ -1,201 +0,0 @@
|
|||||||
/*
|
|
||||||
* 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 ${groupId}.${rootArtifactId}.api;
|
|
||||||
|
|
||||||
import java.util.ArrayList;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.concurrent.ConcurrentHashMap;
|
|
||||||
|
|
||||||
import ${groupId}.${rootArtifactId}.api.util.APIUtil;
|
|
||||||
import ${groupId}.${rootArtifactId}.plugin.constants.DeviceTypeConstants;
|
|
||||||
import ${groupId}.${rootArtifactId}.api.dto.DeviceJSON;
|
|
||||||
import ${groupId}.${rootArtifactId}.api.transport.MQTTConnector;
|
|
||||||
import ${groupId}.${rootArtifactId}.api.dto.SensorRecord;
|
|
||||||
|
|
||||||
import org.apache.commons.logging.Log;
|
|
||||||
import org.apache.commons.logging.LogFactory;
|
|
||||||
|
|
||||||
import org.wso2.carbon.analytics.dataservice.commons.SortByField;
|
|
||||||
import org.wso2.carbon.analytics.datasource.commons.exception.AnalyticsException;
|
|
||||||
import org.wso2.carbon.analytics.dataservice.commons.SORT;
|
|
||||||
import org.wso2.carbon.apimgt.annotations.api.API;
|
|
||||||
import org.wso2.carbon.device.mgt.common.DeviceIdentifier;
|
|
||||||
import org.wso2.carbon.device.mgt.common.authorization.DeviceAccessAuthorizationException;
|
|
||||||
import org.wso2.carbon.device.mgt.extensions.feature.mgt.annotations.DeviceType;
|
|
||||||
import org.wso2.carbon.device.mgt.extensions.feature.mgt.annotations.Feature;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.controlqueue.mqtt.MqttConfig;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.service.IoTServerStartupListener;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.transport.TransportHandlerException;
|
|
||||||
|
|
||||||
import javax.servlet.http.HttpServletResponse;
|
|
||||||
import javax.ws.rs.Consumes;
|
|
||||||
import javax.ws.rs.FormParam;
|
|
||||||
import javax.ws.rs.QueryParam;
|
|
||||||
import javax.ws.rs.PathParam;
|
|
||||||
import javax.ws.rs.GET;
|
|
||||||
import javax.ws.rs.HeaderParam;
|
|
||||||
import javax.ws.rs.POST;
|
|
||||||
import javax.ws.rs.Path;
|
|
||||||
import javax.ws.rs.Produces;
|
|
||||||
import javax.ws.rs.core.Context;
|
|
||||||
import javax.ws.rs.core.MediaType;
|
|
||||||
import javax.ws.rs.core.Response;
|
|
||||||
|
|
||||||
|
|
||||||
/**
|
|
||||||
* This is the controller API which is used to control agent side functionality
|
|
||||||
*/
|
|
||||||
@SuppressWarnings("NonJaxWsWebServices")
|
|
||||||
@API(name = "${deviceType}", version = "1.0.0", context = "/${deviceType}", tags = "${deviceType}")
|
|
||||||
@DeviceType(value = "${deviceType}")
|
|
||||||
public class ControllerServiceImpl implements ControllerService{
|
|
||||||
|
|
||||||
private static Log log = LogFactory.getLog(ControllerService.class);
|
|
||||||
private MQTTConnector mqttConnector;
|
|
||||||
private ConcurrentHashMap<String, DeviceJSON> deviceToIpMap = new ConcurrentHashMap<>();
|
|
||||||
|
|
||||||
private boolean waitForServerStartup() {
|
|
||||||
while (!IoTServerStartupListener.isServerReady()) {
|
|
||||||
try {
|
|
||||||
Thread.sleep(5000);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
return true;
|
|
||||||
}
|
|
||||||
}
|
|
||||||
return false;
|
|
||||||
}
|
|
||||||
|
|
||||||
public MQTTConnector getMQTTConnector() {
|
|
||||||
return mqttConnector;
|
|
||||||
}
|
|
||||||
|
|
||||||
public void setMQTTConnector(final MQTTConnector MQTTConnector) {
|
|
||||||
Runnable connector = new Runnable() {
|
|
||||||
public void run() {
|
|
||||||
if (waitForServerStartup()) {
|
|
||||||
return;
|
|
||||||
}
|
|
||||||
//The delay is added for the server to starts up.
|
|
||||||
try {
|
|
||||||
Thread.sleep(5000);
|
|
||||||
} catch (InterruptedException e) {
|
|
||||||
Thread.currentThread().interrupt();
|
|
||||||
}
|
|
||||||
ControllerServiceImpl.this.mqttConnector = MQTTConnector;
|
|
||||||
if (MqttConfig.getInstance().isEnabled()) {
|
|
||||||
mqttConnector.connect();
|
|
||||||
} else {
|
|
||||||
log.warn("MQTT disabled in 'devicemgt-config.xml'. Hence, MQTTConnector" +
|
|
||||||
" not started.");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
Thread connectorThread = new Thread(connector);
|
|
||||||
connectorThread.setDaemon(true);
|
|
||||||
connectorThread.start();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @param agentInfo device owner,id and sensor value
|
|
||||||
* @return
|
|
||||||
*/
|
|
||||||
@Path("device/register")
|
|
||||||
@POST
|
|
||||||
@Consumes(MediaType.APPLICATION_JSON)
|
|
||||||
public Response registerDevice(final DeviceJSON agentInfo) {
|
|
||||||
String deviceId = agentInfo.deviceId;
|
|
||||||
if ((agentInfo.deviceId != null) && (agentInfo.owner != null)) {
|
|
||||||
deviceToIpMap.put(deviceId, agentInfo);
|
|
||||||
return Response.status(Response.Status.OK).build();
|
|
||||||
}
|
|
||||||
return Response.status(Response.Status.NOT_ACCEPTABLE).build();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* @param deviceId unique identifier for given device type
|
|
||||||
* @param state change status of sensor: on/off
|
|
||||||
* @param response
|
|
||||||
*/
|
|
||||||
@Path("device/{deviceId}/change-status")
|
|
||||||
@POST
|
|
||||||
@Feature(code = "change-status", name = "Change status of sensor: on/off", type = "operation",
|
|
||||||
description = "Change status of sensor: on/off")
|
|
||||||
public Response changeStatus(@PathParam("deviceId") String deviceId,
|
|
||||||
@QueryParam("state") String state,
|
|
||||||
@Context HttpServletResponse response) {
|
|
||||||
try {
|
|
||||||
if (!APIUtil.getDeviceAccessAuthorizationService().isUserAuthorized(new DeviceIdentifier(deviceId,
|
|
||||||
DeviceTypeConstants.DEVICE_TYPE))) {
|
|
||||||
return Response.status(Response.Status.UNAUTHORIZED.getStatusCode()).build();
|
|
||||||
}
|
|
||||||
String sensorState = state.toUpperCase();
|
|
||||||
if (!sensorState.equals(DeviceTypeConstants.STATE_ON) && !sensorState.equals(
|
|
||||||
DeviceTypeConstants.STATE_OFF)) {
|
|
||||||
log.error("The requested state change should be either - 'ON' or 'OFF'");
|
|
||||||
return Response.status(Response.Status.BAD_REQUEST.getStatusCode()).build();
|
|
||||||
}
|
|
||||||
String mqttResource = DeviceTypeConstants.SENSOR_CONTEXT.replace("/", "");
|
|
||||||
mqttConnector.publishDeviceData(deviceId, mqttResource, sensorState);
|
|
||||||
return Response.ok().build();
|
|
||||||
} catch (TransportHandlerException e) {
|
|
||||||
log.error("Failed to send switch-bulb request to device [" + deviceId + "]");
|
|
||||||
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
|
|
||||||
} catch (DeviceAccessAuthorizationException e) {
|
|
||||||
log.error(e.getErrorMessage(), e);
|
|
||||||
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Retrieve Sensor data for the ${deviceType}
|
|
||||||
*/
|
|
||||||
@Path("device/stats/{deviceId}")
|
|
||||||
@GET
|
|
||||||
@Consumes("application/json")
|
|
||||||
@Produces("application/json")
|
|
||||||
public Response getSensorStats(@PathParam("deviceId") String deviceId, @QueryParam("from") long from,
|
|
||||||
@QueryParam("to") long to){
|
|
||||||
String fromDate = String.valueOf(from);
|
|
||||||
String toDate = String.valueOf(to);
|
|
||||||
String query = "deviceId:" + deviceId + " AND deviceType:" +
|
|
||||||
DeviceTypeConstants.DEVICE_TYPE + " AND time : [" + fromDate + " TO " + toDate + "]";
|
|
||||||
String sensorTableName = DeviceTypeConstants.TEMPERATURE_EVENT_TABLE;
|
|
||||||
try {
|
|
||||||
if (!APIUtil.getDeviceAccessAuthorizationService().isUserAuthorized(new DeviceIdentifier(deviceId,
|
|
||||||
DeviceTypeConstants.DEVICE_TYPE))) {
|
|
||||||
return Response.status(Response.Status.UNAUTHORIZED.getStatusCode()).build();
|
|
||||||
}
|
|
||||||
if (sensorTableName != null) {
|
|
||||||
List<SortByField> sortByFields = new ArrayList<>();
|
|
||||||
SortByField sortByField = new SortByField("time", SORT.ASC, false);
|
|
||||||
sortByFields.add(sortByField);
|
|
||||||
List<SensorRecord> sensorRecords = APIUtil.getAllEventsForDevice(sensorTableName, query, sortByFields);
|
|
||||||
return Response.status(Response.Status.OK.getStatusCode()).entity(sensorRecords).build();
|
|
||||||
}
|
|
||||||
} catch (AnalyticsException e) {
|
|
||||||
String errorMsg = "Error on retrieving stats on table " + sensorTableName + " with query " + query;
|
|
||||||
log.error(errorMsg);
|
|
||||||
return Response.status(Response.Status.INTERNAL_SERVER_ERROR.getStatusCode()).entity(errorMsg).build();
|
|
||||||
} catch (DeviceAccessAuthorizationException e) {
|
|
||||||
log.error(e.getErrorMessage(), e);
|
|
||||||
return Response.status(Response.Status.INTERNAL_SERVER_ERROR).build();
|
|
||||||
}
|
|
||||||
return Response.status(Response.Status.BAD_REQUEST).build();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
57
modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/ControllerService.java → modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/DeviceTypeService.java
57
modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/ControllerService.java → modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/DeviceTypeService.java
197
modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/ManagerServiceImpl.java → modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/DeviceTypeServiceImpl.java
197
modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/ManagerServiceImpl.java → modules/tools/cdmf-devicetype-archetype/src/main/resources/archetype-resources/component/api/src/main/java/__groupId__/__rootArtifactId__/api/DeviceTypeServiceImpl.java
@ -1,56 +0,0 @@
|
|||||||
/*
|
|
||||||
* 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 ${groupId}.${rootArtifactId}.api;
|
|
||||||
|
|
||||||
import org.wso2.carbon.apimgt.annotations.api.API;
|
|
||||||
|
|
||||||
import javax.ws.rs.*;
|
|
||||||
import javax.ws.rs.core.MediaType;
|
|
||||||
import javax.ws.rs.core.Response;
|
|
||||||
|
|
||||||
@Path("enrollment")
|
|
||||||
@API(name = "${deviceType}_mgt", version = "1.0.0", context = "/${deviceType}_mgt", tags = "${deviceType}")
|
|
||||||
public interface ManagerService {
|
|
||||||
|
|
||||||
@Path("/devices/{device_id}")
|
|
||||||
@DELETE
|
|
||||||
Response removeDevice(@PathParam("device_id") String deviceId);
|
|
||||||
|
|
||||||
@Path("/devices/{device_id}")
|
|
||||||
@PUT
|
|
||||||
Response updateDevice(@PathParam("device_id") String deviceId, @QueryParam("name") String name);
|
|
||||||
|
|
||||||
@Path("/devices/{device_id}")
|
|
||||||
@GET
|
|
||||||
@Consumes(MediaType.APPLICATION_JSON)
|
|
||||||
@Produces(MediaType.APPLICATION_JSON)
|
|
||||||
Response getDevice(@PathParam("device_id") String deviceId);
|
|
||||||
|
|
||||||
@Path("/devices")
|
|
||||||
@GET
|
|
||||||
@Consumes(MediaType.APPLICATION_JSON)
|
|
||||||
@Produces(MediaType.APPLICATION_JSON)
|
|
||||||
Response getAllDevices();
|
|
||||||
|
|
||||||
@Path("/devices/download")
|
|
||||||
@GET
|
|
||||||
@Produces("application/zip")
|
|
||||||
Response downloadSketch(@QueryParam("deviceName") String deviceName, @QueryParam("sketchType") String sketchType);
|
|
||||||
|
|
||||||
}
|
|
||||||
@ -1,211 +0,0 @@
|
|||||||
/*
|
|
||||||
* 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 ${groupId}.${rootArtifactId}.api.transport;
|
|
||||||
|
|
||||||
import java.nio.charset.StandardCharsets;
|
|
||||||
|
|
||||||
import ${groupId}.${rootArtifactId}.api.util.APIUtil;
|
|
||||||
import ${groupId}.${rootArtifactId}.plugin.constants.DeviceTypeConstants;
|
|
||||||
import org.apache.commons.logging.Log;
|
|
||||||
import org.apache.commons.logging.LogFactory;
|
|
||||||
import org.eclipse.paho.client.mqttv3.MqttException;
|
|
||||||
import org.eclipse.paho.client.mqttv3.MqttMessage;
|
|
||||||
import org.wso2.carbon.apimgt.application.extension.APIManagementProviderService;
|
|
||||||
import org.wso2.carbon.apimgt.application.extension.dto.ApiApplicationKey;
|
|
||||||
import org.wso2.carbon.apimgt.application.extension.exception.APIManagerException;
|
|
||||||
import org.wso2.carbon.context.PrivilegedCarbonContext;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.controlqueue.mqtt.MqttConfig;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.transport.TransportHandlerException;
|
|
||||||
import org.wso2.carbon.device.mgt.iot.transport.mqtt.MQTTTransportHandler;
|
|
||||||
import org.wso2.carbon.identity.jwt.client.extension.JWTClient;
|
|
||||||
import org.wso2.carbon.identity.jwt.client.extension.dto.AccessTokenInfo;
|
|
||||||
import org.wso2.carbon.identity.jwt.client.extension.exception.JWTClientException;
|
|
||||||
import org.wso2.carbon.user.api.UserStoreException;
|
|
||||||
|
|
||||||
import java.nio.charset.StandardCharsets;
|
|
||||||
import java.util.UUID;
|
|
||||||
|
|
||||||
|
|
||||||
import java.io.File;
|
|
||||||
import java.util.Calendar;
|
|
||||||
import java.util.UUID;
|
|
||||||
|
|
||||||
/**
|
|
||||||
* MQTT is used as transport protocol. So this will provide basic functional requirement in order to communicate over
|
|
||||||
* MQTT
|
|
||||||
*/
|
|
||||||
@SuppressWarnings("no JAX-WS annotation")
|
|
||||||
public class MQTTConnector extends MQTTTransportHandler {
|
|
||||||
|
|
||||||
private static final String publisherContext = "publisher";
|
|
||||||
private static final String subscriberContext = "subscriber";
|
|
||||||
private static final String subscribeTopic = "wso2/+/"+ DeviceTypeConstants.DEVICE_TYPE + "/+/" + publisherContext;
|
|
||||||
private static final String KEY_TYPE = "PRODUCTION";
|
|
||||||
private static final String EMPTY_STRING = "";
|
|
||||||
private static final String JSON_SERIAL_KEY = "SerialNumber";
|
|
||||||
private static final String JSON_TENANT_KEY = "Tenant";
|
|
||||||
private static Log log = LogFactory.getLog(MQTTConnector.class);
|
|
||||||
private static String iotServerSubscriber = UUID.randomUUID().toString().substring(0, 5);
|
|
||||||
|
|
||||||
private MQTTConnector() {
|
|
||||||
super(iotServerSubscriber, DeviceTypeConstants.DEVICE_TYPE,
|
|
||||||
MqttConfig.getInstance().getMqttQueueEndpoint(), subscribeTopic);
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* {@inheritDoc}
|
|
||||||
* This method will initialize connection with message broker
|
|
||||||
*/
|
|
||||||
@Override
|
|
||||||
public void connect() {
|
|
||||||
Runnable connector = new Runnable() {
|
|
||||||
public void run() {
|
|
||||||
while (!isConnected()) {
|
|
||||||
PrivilegedCarbonContext.startTenantFlow();
|
|
||||||
PrivilegedCarbonContext.getThreadLocalCarbonContext().setTenantDomain(
|
|
||||||
DeviceTypeConstants.DEVICE_TYPE_PROVIDER_DOMAIN, true);
|
|
||||||
try {
|
|
||||||
String applicationUsername = PrivilegedCarbonContext.getThreadLocalCarbonContext().getUserRealm()
|
|
||||||
.getRealmConfiguration().getAdminUserName();
|
|
||||||
PrivilegedCarbonContext.getThreadLocalCarbonContext().setUsername(applicationUsername);
|
|
||||||
APIManagementProviderService apiManagementProviderService = APIUtil
|
|
||||||
.getAPIManagementProviderService();
|
|
||||||
String[] tags = {DeviceTypeConstants.DEVICE_TYPE};
|
|
||||||
ApiApplicationKey apiApplicationKey = apiManagementProviderService
|
|
||||||
.generateAndRetrieveApplicationKeys(DeviceTypeConstants.DEVICE_TYPE, tags, KEY_TYPE,
|
|
||||||
applicationUsername, true);
|
|
||||||
JWTClient jwtClient = APIUtil.getJWTClientManagerService().getJWTClient();
|
|
||||||
String scopes = "device_type_" + DeviceTypeConstants.DEVICE_TYPE + " device_mqtt_connector";
|
|
||||||
AccessTokenInfo accessTokenInfo = jwtClient.getAccessToken(apiApplicationKey.getConsumerKey(),
|
|
||||||
apiApplicationKey.getConsumerSecret(), applicationUsername, scopes);
|
|
||||||
//create token
|
|
||||||
String accessToken = accessTokenInfo.getAccessToken();
|
|
||||||
setUsernameAndPassword(accessToken, EMPTY_STRING);
|
|
||||||
connectToQueue();
|
|
||||||
} catch (TransportHandlerException e) {
|
|
||||||
log.error("Connection/Subscription to MQTT Broker at: " + mqttBrokerEndPoint + " failed", e);
|
|
||||||
try {
|
|
||||||
Thread.sleep(timeoutInterval);
|
|
||||||
} catch (InterruptedException ex) {
|
|
||||||
log.error("MQTT-Connector: Thread Sleep Interrupt Exception.", ex);
|
|
||||||
}
|
|
||||||
} catch (JWTClientException e) {
|
|
||||||
log.error("Failed to retrieve token from JWT Client.", e);
|
|
||||||
return;
|
|
||||||
} catch (UserStoreException e) {
|
|
||||||
log.error("Failed to retrieve the user.", e);
|
|
||||||
return;
|
|
||||||
} catch (APIManagerException e) {
|
|
||||||
log.error("Failed to create an application and generate keys.", e);
|
|
||||||
return;
|
|
||||||
} finally {
|
|
||||||
PrivilegedCarbonContext.endTenantFlow();
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
Thread connectorThread = new Thread(connector);
|
|
||||||
connectorThread.start();
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* This callback function will be called by message broker when some messages available to subscribed topic
|
|
||||||
*
|
|
||||||
* @param message mqtt message which is comming form agent side
|
|
||||||
* @param messageParams metadata of mqtt message
|
|
||||||
*/
|
|
||||||
@Override
|
|
||||||
public void processIncomingMessage(MqttMessage message, String... messageParams) throws TransportHandlerException {
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* connection with message broker can be terminated
|
|
||||||
*/
|
|
||||||
@Override
|
|
||||||
public void disconnect() {
|
|
||||||
Runnable stopConnection = new Runnable() {
|
|
||||||
public void run() {
|
|
||||||
while (isConnected()) {
|
|
||||||
try {
|
|
||||||
closeConnection();
|
|
||||||
} catch (MqttException e) {
|
|
||||||
if (log.isDebugEnabled()) {
|
|
||||||
log.warn("Unable to 'STOP' MQTT connection at broker at: " + mqttBrokerEndPoint);
|
|
||||||
}
|
|
||||||
try {
|
|
||||||
Thread.sleep(timeoutInterval);
|
|
||||||
} catch (InterruptedException e1) {
|
|
||||||
log.error("MQTT-Terminator: Thread Sleep Interrupt Exception");
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
};
|
|
||||||
Thread terminatorThread = new Thread(stopConnection);
|
|
||||||
terminatorThread.setDaemon(true);
|
|
||||||
terminatorThread.start();
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void publishDeviceData() throws TransportHandlerException {
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void publishDeviceData(MqttMessage publishData) throws TransportHandlerException {
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void publishDeviceData(String... publishData) throws TransportHandlerException {
|
|
||||||
if (publishData.length != 3) {
|
|
||||||
String errorMsg = "Incorrect number of arguments received to SEND-MQTT Message. " +
|
|
||||||
"Need to be [owner, deviceId, resource{SENSOR}, state{ON/OFF or null}]";
|
|
||||||
log.error(errorMsg);
|
|
||||||
throw new TransportHandlerException(errorMsg);
|
|
||||||
}
|
|
||||||
|
|
||||||
String deviceId = publishData[0];
|
|
||||||
String resource = publishData[1];
|
|
||||||
String state = publishData[2];
|
|
||||||
|
|
||||||
MqttMessage pushMessage = new MqttMessage();
|
|
||||||
String publishTopic = "wso2/" + APIUtil.getTenantDomainOftheUser() + "/" + DeviceTypeConstants.DEVICE_TYPE
|
|
||||||
+ "/" + deviceId;
|
|
||||||
String actualMessage = resource + ":" + state;
|
|
||||||
pushMessage.setPayload(actualMessage.getBytes(StandardCharsets.UTF_8));
|
|
||||||
pushMessage.setQos(DEFAULT_MQTT_QUALITY_OF_SERVICE);
|
|
||||||
pushMessage.setRetained(false);
|
|
||||||
publishToQueue(publishTopic, pushMessage);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void processIncomingMessage() {
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void processIncomingMessage(MqttMessage message) throws TransportHandlerException {
|
|
||||||
|
|
||||||
}
|
|
||||||
|
|
||||||
}
|
|
||||||
Loading…
Reference in new issue