community
/
device-mgt-plugins
10
0
Fork 39
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Fix for default policy updates in virtual fire alarm

Browse Source
merge-requests/1/head
Nirothipan 7 years ago
parent a17b8a16ec
commit 6cfe4ac3e7
4 changed files with 35 additions and 30 deletions
Show Stats Download Patch File Download Diff File

1
components/device-types/virtual-fire-alarm-plugin/org.wso2.carbon.device.mgt.iot.virtualfirealarm.agent.advanced.impl/src/main/java/org/wso2/carbon/device/mgt/iot/virtualfirealarm/agent/advanced/communication/mqtt/FireAlarmMQTTCommunicator.java
Unescape View File

@ -176,6 +176,7 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
} else {
JSONObject jsonMessage = new JSONObject(receivedMessage);
updateCEPPolicy(jsonMessage.getString("policyDefinition"));
}
log.info(AgentConstants.LOG_APPENDER + "Message [" + receivedMessage + "] was received");
} catch (AgentCoreOperationException e) {

4
components/device-types/virtual-fire-alarm-plugin/org.wso2.carbon.device.mgt.iot.virtualfirealarm.agent.advanced.impl/src/main/java/org/wso2/carbon/device/mgt/iot/virtualfirealarm/agent/advanced/core/AgentManager.java
Unescape View File

@ -125,8 +125,8 @@ public class AgentManager {
agentCommunicator.put(AgentConstants.XMPP_PROTOCOL, xmppCommunicator);
} catch (TransportHandlerException e) {
log.error("XMPP Endpoint String - " + agentConfigs.getXmppServerEndpoint() +
", provided in the configuration file is invalid.");
log.info("XMPP Endpoint String - " + agentConfigs.getXmppServerEndpoint() +
", provided in the configuration file is invalid. XMPP is not configured.");
}
String mqttTopic = String.format(AgentConstants.MQTT_SUBSCRIBE_TOPIC, agentConfigs.getTenantDomain(),
agentConfigs.getDeviceId());

56
components/device-types/virtual-fire-alarm-plugin/org.wso2.carbon.device.mgt.iot.virtualfirealarm.agent.advanced.impl/src/main/java/org/wso2/carbon/device/mgt/iot/virtualfirealarm/agent/advanced/sidhdhi/SidhdhiQuery.java
Unescape View File

@ -68,32 +68,34 @@ public class SidhdhiQuery implements Runnable {
//Start the execution plan with pre-defined or previously persisted Siddhi query
File f = new File(sidhdhiQueryPath);
while (true) {
if (!f.exists()) {
AgentUtilOperations.writeToFile("", sidhdhiQueryPath);
}
if (f.exists()) {
//AgentUtilOperations.writeToFile("", sidhdhiQueryPath);
StartExecutionPlan startExecutionPlan = new StartExecutionPlan().invoke();
while (true) {
StartExecutionPlan startExecutionPlan = new StartExecutionPlan().invoke();
//Check if there is new policy update available
if (AgentManager.isUpdated()) {
System.out.print("### Policy Update Detected!");
//Restart execution plan with new query
restartSiddhi();
startExecutionPlan = new StartExecutionPlan().invoke();
}
InputHandler inputHandler = startExecutionPlan.getInputHandler();
//Sending events to Siddhi
try {
int humidityReading = AgentManager.getInstance().getTemperature();
inputHandler.send(new Object[]{"FIRE_1", humidityReading});
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
break;
//Check if there is new policy update available
if (AgentManager.isUpdated()) {
System.out.print("### Policy Update Detected!");
//Restart execution plan with new query
restartSiddhi();
startExecutionPlan = new StartExecutionPlan().invoke();
}
InputHandler inputHandler = startExecutionPlan.getInputHandler();
//Sending events to Siddhi
try {
int humidityReading = AgentManager.getInstance().getTemperature();
inputHandler.send(new Object[]{"FIRE_1", humidityReading});
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
break;
}
}
}
}
@ -117,7 +119,9 @@ public class SidhdhiQuery implements Runnable {
public static String readFile(String path, Charset encoding) {
byte[] encoded = new byte[0];
try {
encoded = Files.readAllBytes(Paths.get(path));
if(new File(sidhdhiQueryPath).exists()){
encoded = Files.readAllBytes(Paths.get(path));
}
} catch (IOException e) {
log.error("Error reading Sidhdhi query from file.");
}
@ -172,7 +176,7 @@ public class SidhdhiQuery implements Runnable {
siddhiManager.addCallback("bulbOnStream", new StreamCallback() {
@Override
public void receive(Event[] events) {
System.out.println("Bulb on Event Fired!");
// System.out.println("Bulb on Event Fired!");
if (events.length > 0) {
if (!AgentManager.getInstance().isAlarmOn()) {
AgentManager.getInstance().changeAlarmStatus(true);
@ -185,7 +189,7 @@ public class SidhdhiQuery implements Runnable {
siddhiManager.addCallback("bulbOffStream", new StreamCallback() {
@Override
public void receive(Event[] inEvents) {
System.out.println("Bulb off Event Fired");
// System.out.println("Bulb off Event Fired");
if (AgentManager.getInstance().isAlarmOn()) {
AgentManager.getInstance().changeAlarmStatus(false);
System.out.println("#### Performed HTTP call! OFF.");
@ -198,7 +202,7 @@ public class SidhdhiQuery implements Runnable {
inputHandler = siddhiManager.getInputHandler("fireAlarmEventStream");
//Starting event processing
System.out.println("Execution Plan Started!");
// System.out.println("Execution Plan Started!");
return this;
}
}

4
components/device-types/virtual-fire-alarm-plugin/org.wso2.carbon.device.mgt.iot.virtualfirealarm.agent.impl/src/main/java/org/wso2/carbon/device/mgt/iot/virtualfirealarm/agent/core/AgentManager.java
Unescape View File

@ -114,8 +114,8 @@ public class AgentManager {
agentCommunicator.put(AgentConstants.XMPP_PROTOCOL, xmppCommunicator);
} catch (TransportHandlerException e) {
log.error("XMPP Endpoint String - " + agentConfigs.getXmppServerEndpoint() +
", provided in the configuration file is invalid.");
log.info("XMPP Endpoint String - " + agentConfigs.getXmppServerEndpoint() +
", provided in the configuration file is invalid. XMPP is not configured.");
}
String mqttTopic = String.format(AgentConstants.MQTT_SUBSCRIBE_TOPIC, agentConfigs.getTenantDomain(),
agentConfigs.getDeviceId());

Write Preview
Loading…
Cancel
Save