|
|
@ -85,7 +85,7 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
|
|
|
|
|
|
|
|
|
|
|
|
} catch (TransportHandlerException e) {
|
|
|
|
} catch (TransportHandlerException e) {
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "Subscription to MQTT Broker at: " +
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "Subscription to MQTT Broker at: " +
|
|
|
|
mqttBrokerEndPoint + " failed");
|
|
|
|
mqttBrokerEndPoint + " failed");
|
|
|
|
agentManager.updateAgentStatus("Subscription to broker failed.");
|
|
|
|
agentManager.updateAgentStatus("Subscription to broker failed.");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
@ -118,16 +118,6 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
|
|
|
|
receivedMessage = message.toString();
|
|
|
|
receivedMessage = message.toString();
|
|
|
|
if(!receivedMessage.contains("policyDefinition")){
|
|
|
|
if(!receivedMessage.contains("policyDefinition")){
|
|
|
|
receivedMessage = AgentUtilOperations.extractMessageFromPayload(receivedMessage);
|
|
|
|
receivedMessage = AgentUtilOperations.extractMessageFromPayload(receivedMessage);
|
|
|
|
}else{
|
|
|
|
|
|
|
|
JSONObject jsonMessage = new JSONObject(receivedMessage);
|
|
|
|
|
|
|
|
updateCEPPolicy(jsonMessage.getString("policyDefinition"));
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
log.info(AgentConstants.LOG_APPENDER + "Message [" + receivedMessage + "] was received");
|
|
|
|
|
|
|
|
} catch (AgentCoreOperationException e) {
|
|
|
|
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "Could not extract message from payload.", e);
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
String[] controlSignal = receivedMessage.split(":");
|
|
|
|
String[] controlSignal = receivedMessage.split(":");
|
|
|
|
// message- "<SIGNAL_TYPE>:<SIGNAL_MODE>" format.(ex: "BULB:ON", "TEMPERATURE", "HUMIDITY")
|
|
|
|
// message- "<SIGNAL_TYPE>:<SIGNAL_MODE>" format.(ex: "BULB:ON", "TEMPERATURE", "HUMIDITY")
|
|
|
@ -168,9 +158,8 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
|
|
|
|
publishToQueue(humidPublishTopic, securePayLoad);
|
|
|
|
publishToQueue(humidPublishTopic, securePayLoad);
|
|
|
|
break;
|
|
|
|
break;
|
|
|
|
|
|
|
|
|
|
|
|
case AgentConstants.POLICY_SIGNAL:
|
|
|
|
case AgentConstants.POLICY_REVOKE:
|
|
|
|
String policy = controlSignal[1];
|
|
|
|
break;
|
|
|
|
updateCEPPolicy(policy);
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
default:
|
|
|
|
default:
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "'" + controlSignal[0] +
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "'" + controlSignal[0] +
|
|
|
@ -185,6 +174,17 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
|
|
|
|
agentManager.getAgentConfigs().getMqttBrokerEndpoint() + " failed");
|
|
|
|
agentManager.getAgentConfigs().getMqttBrokerEndpoint() + " failed");
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
} else {
|
|
|
|
|
|
|
|
JSONObject jsonMessage = new JSONObject(receivedMessage);
|
|
|
|
|
|
|
|
updateCEPPolicy(jsonMessage.getString("policyDefinition"));
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
log.info(AgentConstants.LOG_APPENDER + "Message [" + receivedMessage + "] was received");
|
|
|
|
|
|
|
|
} catch (AgentCoreOperationException e) {
|
|
|
|
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER + "Could not extract message from payload.", e);
|
|
|
|
|
|
|
|
return;
|
|
|
|
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
@Override
|
|
|
|
@Override
|
|
|
@ -247,8 +247,8 @@ public class FireAlarmMQTTCommunicator extends MQTTTransportHandler {
|
|
|
|
} catch (MqttException e) {
|
|
|
|
} catch (MqttException e) {
|
|
|
|
if (log.isDebugEnabled()) {
|
|
|
|
if (log.isDebugEnabled()) {
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER +
|
|
|
|
log.warn(AgentConstants.LOG_APPENDER +
|
|
|
|
"Unable to 'STOP' MQTT connection at broker at: " +
|
|
|
|
"Unable to 'STOP' MQTT connection at broker at: " +
|
|
|
|
mqttBrokerEndPoint);
|
|
|
|
mqttBrokerEndPoint);
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
|
|
|
|
try {
|
|
|
|
try {
|
|
|
|