PIOT-GDA-11-004: Implement the data capture and event triggering logic between the GDA and cloud service

[labbenchstudios]

Description

• Implement the requisite services and functionality within your selected cloud service environment to do the following:
  • Collect and store data: Within the cloud service, collect sensor data from the CDA and system performance data from both the CDA and GDA via the appropriately named topics.
  • Configure LED actuation events: Within the cloud service, create a rule that analyzes the sensor or system performance data for threshold crossings you choose. On threshold crossing, generate an actuation event to trigger an LED ON or OFF and write that to an LED actuation event topic.
  • Connect the GDA to the LED actuation event topic: In the GDA's CloudClientConnector implementation, subscribe to the LED actuation event topic after successful connection. Make sure the cloud service has created the LED actuation event topic prior to the subscribe request. Use a callback handler class to handle any incoming LED actuation events.

Objective

• Show the following
  • Connect the CDA to the GDA and the GDA to the cloud service
  • Send data from the CDA to the GDA and from the GDA to the cloud service
  • Send at least one actuation event from the cloud service to the GDA based on the data the cloud service receives from the CDA via the GDA
  • Use the cloud service actuation event received by the GDA to trigger an actuation event on the CDA that can be observed (e.g., generate a log message, activate the LED on the Sense HAT emulator, or illuminate the LED display on the Sense HAT device)
  • Verify end to end functionality and show screenshots of the results in your README.md, as follows:
    • At least three variables collected and stored in the cloud service (e.g., memory utilization, CPU utilization, temperature, etc.)
    • At least one actuation event triggered in the cloud service (e.g., LED transitioning from OFF to ON, or ON to OFF)
    • At least one actuation event where the CDA receives and processes the actuation event triggered in the cloud service (e.g., LED 'ON' actuation event triggered in the cloud, received by the GDA, and then sent to the CDA for processing via a log message, the Sense HAT emulator, or the Sense HAT device)

Review the README

• Please see README.md for further information on, and use of, this content.
• License for embedded documentation and source codes: PIOT-DOC-LIC

Estimated effort may vary greatly

• The estimated level of effort for this exercise shown in the 'Estimate' section below is a very rough approximation. The actual level of effort may vary greatly depending on your development and test environment, experience with the requisite technologies, and many other factors.

Actions

These Updates Span the Cloud Service, GDA, and the CDA

• Within the cloud service, create a device and variable representing the LED actuation event topic.
• Within the GDA, subscribe to the cloud service's LED actuation event topic that the cloud service will use for publishing its actuation events (as referenced in the Description).
  • NOTE: Make sure the name of the topic the cloud service is using and the name of the topic you subscribe to below are exactly the same.
  • You can implement the subscribe in one of three places:
    • CloudClientConnector after successful connection
    • MqttClientConnector after successful connection: if useCloudGatewayConfig is true, this could be done within connectComplete()
    • DeviceDataManager after successful connection of CloudClientConnector
• Within the GDA, upon receipt of the LED actuation event, do the following:
  • Generate an ActuatorData instance for the CDA, and set the state data to 'ON' or 'OFF' depending on the value received (e.g., 0 == 'ON'; 1 == 'OFF'). Pass the ActuatorData instance OR its JSON conversion to DeviceDataManager for processing and passing to the CDA.
    • It's important to note that the cloud service will need to generate a value of '1' (ON) or a value of '0' (OFF) to use this parsing approach. Remember that the CDA will eventually receive this actuator command and will have to parse the value to determine what action it takes (e.g., turn the LED display on the Sense HAT Emulator 'ON' or 'OFF' based on the value sent from the cloud).
    • There are many ways to implement this functionality - one to consider is to simply invoke the IDataMessageListener implementation's (DeviceDataManager) handleIncomingMessage(ResourceNameEnum resourceName, String msg) callback once the MQTT message is received, and simply update the logic of of the handleIncomingMessage() method to parse the data and reconstitute the ActuatorData instance from the JSON.
  • After parsing the ActuatorData message in handleIncomingMessage() and reconstituting the ActuatorData instance, pass the data onto the CDA's actuator command topic via MQTT or CoAP.
• Within the CDA, verify the logic is implemented to receive the ActuatorData instance and hand it off to ActuatorAdapterManager. This functionality should already be in place, so you shouldn't need to add any logic to the CDA for this lab module.

GDA Implementation Notes

NOTE: Here's one approach for implementing the logic needed for the GDA and cloud service to work together. There are many other ways to design your solution. The sample code shown for CloudClientConnector was tested with Ubidots as the cloud service provider.

Create IConnectionListener

NOTE: The IConnectionListener interface should already be part of the java-components source repository.

• IConnectionListener: Add the IConnectionListener interface to the programmingtheiot.gda.connection package. This will be implemented by CloudClientConnector and used by the MqttClientConnector to notify CloudClientConnector when key connection events complete.

package programmingtheiot.gda.connection;

public interface IConnectionListener
{
	public void onConnect();
	
	public void onDisconnect();
}

Make Sure MqttClientConnector is Updated to Use IConnectionListener

NOTE: These updates should already have been implemented as part of PIOT-GDA-11-001.

• MqttClientConnector: Within MqttClientConnector, add a class-scoped reference for IConnectionListener and add the setConnectionListener(IConnectionListener listener) method for setting the IConnectionListener instance.

// TODO: place this in the class-scoped variable declarations section
private IConnectionListener connListener = null;

• MqttClientConnector: Within MqttClientConnector, add the setConnectionListener(IConnectionListener listener) method for setting the IConnectionListener instance.

public boolean setConnectionListener(IConnectionListener listener)
{
	if (listener != null) {
		_Logger.info("Setting connection listener.");
		this.connListener = listener;
		return true;
	} else {
		_Logger.warning("No connection listener specified. Ignoring.");
	}
	
	return false;
}

• MqttClientConnector: Within MqttClientConnector's connectComplete() callback method, add a call to the IConnectionListener instances onConnect() method.
  • NOTE: Your implementation may look similar to the following, but even if slightly different, the important addition here is to the onConnect() call.

@Override
public void connectComplete(boolean reconnect, String serverURI)
{
	_Logger.info("MQTT connection successful (is reconnect = " + reconnect + "). Broker: " + serverURI);
	
	int qos = 1;
	
	try {
		if (! this.useCloudGatewayConfig) {
			_Logger.info("Subscribing to topic: " + ResourceNameEnum.CDA_ACTUATOR_RESPONSE_RESOURCE.getResourceName());
			
			this.mqttClient.subscribe(
				ResourceNameEnum.CDA_ACTUATOR_RESPONSE_RESOURCE.getResourceName(),
				qos,
				new ActuatorResponseMessageListener(ResourceNameEnum.CDA_ACTUATOR_RESPONSE_RESOURCE, this.dataMsgListener));
		}
	} catch (MqttException e) {
		_Logger.warning("Failed to subscribe to CDA actuator response topic.");
	}
	
	// This is the new addition - a call to this.connListener.onConnect()
	if (this.connListener != null) {
		this.connListener.onConnect();
	}
}

Update CloudClientConnector

• CloudClientConnector: Implement a callback handler to process incoming LED actuation events from the cloud service within CloudClientConnector. This can be implemented as an inner class.
  • IMPORTANT NOTE: You can implement multiple IMqttMessageListener inner classes and subscribe to additional topics with each IMqttMessageListener implementation if needed. The example below is provided as a template for consideration.

private class LedEnablementMessageListener implements IMqttMessageListener
{
	private IDataMessageListener dataMsgListener = null;
	
	private ResourceNameEnum resource = ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE;
	
	private int    typeID   = ConfigConst.LED_ACTUATOR_TYPE;
	private String itemName = ConfigConst.LED_ACTUATOR_NAME;
	
	LedEnablementMessageListener(IDataMessageListener dataMsgListener)
	{
		this.dataMsgListener = dataMsgListener;
	}
	
	public ResourceNameEnum getResource()
	{
		return this.resource;
	}
	
	@Override
	public void messageArrived(String topic, MqttMessage message) throws Exception
	{
		try {
			String jsonData = new String(message.getPayload());
			
			ActuatorData actuatorData =
				DataUtil.getInstance().jsonToActuatorData(jsonData);
			
			// TODO: This will have to match the CDA's location ID, depending on the
			// validation logic implemented within the CDA's ActuatorAdapterManager
			actuatorData.setLocationID(ConfigConst.CONSTRAINED_DEVICE);
			actuatorData.setTypeID(this.typeID);
			actuatorData.setName(this.itemName);
			
			int val = (int) actuatorData.getValue();
			
			switch (val) {
				case ConfigConst.ON_COMMAND:
					_Logger.info("Received LED enablement message [ON].");
					actuatorData.setStateData("LED switching ON");
					break;
					
				case ConfigConst.OFF_COMMAND:
					_Logger.info("Received LED enablement message [OFF].");
					actuatorData.setStateData("LED switching OFF");
					break;
					
				default:
					return;
			}
			
			// There are two ways to handle passing of ActuatorData messages
			// from this method to IDataMessageListener (DeviceDataManager):
			// 
			// Option 1: Pass the JSON payload (which will likely be ActuatorData).
			// Option 2: Pass the ActuatorData instance directly.
			// 
			// The latest version of java-components contains a shell definition
			// for Option 2 (using Actuator Data via handleActuatorCommandRequest()).
			// If you do not have this method defined in IDataMessageListener and
			// DeviceDataManager, you can add it in, or just use Option 1.
			// 
			// Choose which you'd like to use and comment out the other,
			// but DO NOT USE BOTH!
			
			// 
			// Option 1: using JSON
			// 
			if (this.dataMsgListener != null) {
				// NOTE: This conversion is useful for validation purposes and
				// to support the next line of code. You can bypass this if
				// your IDataMessageListener and DeviceDataManager implement:
				// handleActuatorCommandRequest(ActuatorData).

				jsonData = DataUtil.getInstance().actuatorDataToJson(actuatorData);
				
				// NOTE: The implementation of IDataMessageListener, which will be
				// DeviceDataManager, will need to parse the JSON data to handle
				// the actuator command via the handleIncomingMessage() method.
				// The implementation of handleIncomingMessage() will then
				// convert the data back into an ActuatorData instance and
				// send it to the CDA via CoAP or MQTT.
				// 
				// It may seem odd to convert the payload JSON to ActuatorData
				// and then back again to JSON, only to be converted once again
				// to an ActuatorData instance. The purpose of this was originally
				// to support multiple payload types.
				this.dataMsgListener.handleIncomingMessage(
					ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE, jsonData);
			}
			
			// 
			// Option 1: using ActuatorData
			// 
			//if (this.dataMsgListener != null) {
			//	this.dataMsgListener.handleActuatorCommandRequest(
			//		ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE, actuatorData);
			//}
		} catch (Exception e) {
			_Logger.warning("Failed to convert message payload to ActuatorData.");
		}
	}
}

• CloudClientConnector: Implement the IConnectionListener methods within CloudClientConnector - be sure to add implements IConnectionListener to the CloudClientConnector class declaration.

@Override
public void onConnect()
{
	_Logger.info("Handling CSP subscriptions and device topic provisioninig...");
	
	LedEnablementMessageListener ledListener = new LedEnablementMessageListener(this.dataMsgListener);
	
	// topic may not exist yet, so create a 'response' actuation event with invalid value -
	// this will create the relevant topic if it doesn't yet exist, which ensures
	// the message listener (if coded correctly) will log a message but ignore the
	// actuation command and NOT pass it onto the IDataMessageListener instance
	ActuatorData ad = new ActuatorData();
	ad.setAsResponse();
	ad.setName(ConfigConst.LED_ACTUATOR_NAME);
	ad.setValue((float) -1.0); // NOTE: this just needs to be an invalid actuation value

	String ledTopic = createTopicName(leml.getResource().getDeviceName(), ad.getName());
	String adJson = DataUtil.getInstance().actuatorDataToJson(ad);

	this.publishMessageToCloud(ledTopic, adJson);
	
	this.mqttClient.subscribeToTopic(ledTopic, this.qosLevel, ledListener);
}

@Override
public void onDisconnect()
{
	_Logger.info("MQTT client disconnected. Nothing else to do.");
}

• CloudClientConnector: Update connectClient() to pass this to the MqttClientConnector instance as the IConnectionListener implementation.

@Override
public boolean connectClient()
{
	if (this.mqttClient == null) {
		this.mqttClient = new MqttClientConnector(true);
		this.mqttClient.setConnectionListener(this);
	}
	
	return this.mqttClient.connectClient();
}

• CloudClientConnector: For convenience, add in private methods for creating cloud service provider topic names. The examples shown align to Ubidots topic naming conventions, but may work with others as well.

private String createTopicName(ResourceNameEnum resource)
{
	return createTopicName(resource.getDeviceName(), resource.getResourceType());
}

private String createTopicName(ResourceNameEnum resource, String itemName)
{
	return (createTopicName(resource) + "-" + itemName).toLowerCase();
}

private String createTopicName(String deviceName, String resourceTypeName)
{
	StringBuilder buf = new StringBuilder();
	
	if (deviceName != null && deviceName.trim().length() > 0) {
		buf.append(topicPrefix).append(deviceName);
	}
	
	if (resourceTypeName != null && resourceTypeName.trim().length() > 0) {
		buf.append('/').append(resourceTypeName);
	}
	
	return buf.toString().toLowerCase();
}

Update DeviceDataManager

• DeviceDataManager: Within DeviceDataManager, update the handleIncomingMessage(ResourceNameEnum resourceName, String msg) method to process the incoming data (which should be an ActuatorData instance in JSON format).
  • NOTE: Be sure to include some validation within the logic to ensure you're passing only legit ActuatorData commands to the CDA.

@Override
public boolean handleIncomingMessage(ResourceNameEnum resourceName, String msg)
{
	if (resourceName != null && msg != null) {
		try {
			if (resourceName == ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE) {
				_Logger.info("Handling incoming ActuatorData message: " + msg);
				
				// NOTE: it may seem wasteful to convert to ActuatorData and back while
				// the JSON data is already available; however, this provides a validation
				// scheme to ensure the data is actually an 'ActuatorData' instance
				// prior to sending off to the CDA
				ActuatorData ad = DataUtil.getInstance().jsonToActuatorData(msg);
				String jsonData = DataUtil.getInstance().actuatorDataToJson(ad);
				
				if (this.mqttClient != null) {
					// TODO: retrieve the QoS level from the configuration file
					_Logger.fine("Publishing data to MQTT broker: " + jsonData);
					return this.mqttClient.publishMessage(resourceName, jsonData, 0);
				}
				
				// TODO: If the GDA is hosting a CoAP server (or a CoAP client that
				// will connect to the CDA's CoAP server), you can add that logic here
				// in place of the MQTT client or in addition
				
			} else {
				_Logger.warning("Failed to parse incoming message. Unknown type: " + msg);
				
				return false;
			}
		} catch (Exception e) {
			_Logger.log(Level.WARNING, "Failed to process incoming message for resource: " + resourceName, e);
		}
	} else {
		_Logger.warning("Incoming message has no data. Ignoring for resource: " + resourceName);
	}
	
	return false;
}

NOTE: With the updates above, your implementation should have most of what it needs to handle VERY BASIC end-to-end edge to cloud data processing. Your implementation specifics may vary from the notional implementation specified above.

Estimate

• Large

Tests

• In the src/test/java path within the 'programmingtheiot.part04.integration.connection' package, update the CloudClientConnectorTest test class and add the necessary test case methods to execute a full test cycle of the following:
  • Test 1:
    • Create an instance of CloudClientConnector with the appropriate configuration
    • Generate a SensorData sample message
    • Publish the message to the cloud using the CloudClientConnector publish method
    • Verify within the cloud service the data was received and processed
  • Test 2:
    • Create an instance of CloudClientConnector with the appropriate configuration
    • Subscribe to the LED actuator event topic discussed previously, and make sure the cloud service is configured to publish messages to this topic
    • Generate an appropriate number of SensorData sample messages with relevant threshold crossing(s) to trigger at least one actuation event
    • Publish the messages to the cloud using the CloudClientConnector publish method
    • Verify within the cloud service the data was received and processed
    • Verify within the cloud service the actuation event was triggered
    • Verify within CloudClientConnector that the message was received
  • Test 3:
    • Make sure your GDA is configured to start the CloudClientConnector
    • Start the GDA as an application - make sure it will run for at least 5 minutes (or possibly longer), so that the messages generated by the CDA will get processed, sent to the cloud service, and any actuation events will get triggered and sent back from the cloud service to the GDA
    • Start the CDA as an application - make sure it will run for at least 5 minutes (or possibly longer), so that it will generate sufficient messages with threshold crossings that will trigger a cloud-based actuation event
    • The GDA and CDA should communicate via MQTT or CoAP (it's your choice)
    • The GDA and cloud service should communicate via MQTT (for now - you can change this to the cloud service's API for Lab Module 12 if you'd like)
    • After running, you should see end to end message flow from CDA to GDA to cloud service, with actuation events triggered by at least the cloud service, and likely within the GDA and CDA as well (depending on the use case you implement)
    • Sample log output for the GDA (yours will likely be different):

Apr 14, 2022 11:51:31 PM programmingtheiot.gda.connection.MqttClientConnector initCredentialConnectionParameters
INFO: Checking if credentials file exists and us loadable...
.
.
.
Apr 14, 2022 11:51:31 PM programmingtheiot.gda.app.DeviceDataManager startManager
INFO: Successfully connected MQTT client to broker.
Apr 14, 2022 11:51:31 PM programmingtheiot.gda.connection.MqttClientConnector initSecureConnectionParameters
INFO: Configuring TLS...
Apr 14, 2022 11:51:31 PM programmingtheiot.gda.connection.MqttClientConnector initSecureConnectionParameters
.
.
.
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector initSecureConnectionParameters
INFO: TLS enabled.
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector initCredentialConnectionParameters
INFO: Checking if credentials file exists and us loadable...
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector initCredentialConnectionParameters
INFO: Credentials now set.
.
.
.
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.app.DeviceDataManager startManager
INFO: Successfully connected cloud client to CSP.
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector connectComplete
INFO: MQTT connection successful (is reconnect = false). Broker: tcp://localhost:1883
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector connectComplete
INFO: Subscribing to topic: PIOT/ConstrainedDevice/ActuatorResponse
Apr 14, 2022 11:51:32 PM programmingtheiot.gda.connection.MqttClientConnector connectComplete
INFO: MQTT connection successful (is reconnect = false). Broker: .
.
.
.
Apr 14, 2022 11:51:50 PM programmingtheiot.gda.connection.MqttClientConnector deliveryComplete
INFO: Delivered MQTT message with ID: 0
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.connection.CloudClientConnector$LedEnablementMessageListener messageArrived
INFO: Received LED enablement message [OFF].
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.app.DeviceDataManager handleIncomingMessage
INFO: Handling incoming ActuatorData message: {"command":0,"value":0.0,"isResponse":false,"stateData":"LED switching OFF","name":"LedActuator","timeStamp":"2022-04-15T03:51:53.645666900Z","statusCode":0,"typeID":2001,"locationID":"ConstrainedDevice","latitude":0.0,"longitude":0.0,"elevation":0.0,"timeStampMillis":1649994713645}
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.app.DeviceDataManager handleIncomingMessage
INFO: Publishing data to MQTT broker: {"command":0,"value":0.0,"isResponse":false,"stateData":"LED switching OFF","name":"LedActuator","timeStamp":"2022-04-15T03:51:53.645666900Z","statusCode":0,"typeID":2001,"locationID":"ConstrainedDevice","latitude":0.0,"longitude":0.0,"elevation":0.0,"timeStampMillis":1649994713645}
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.connection.MqttClientConnector publishMessage
INFO: Publishing payload value(s) to Ubidots: PIOT/ConstrainedDevice/ActuatorCmd
Payload:
{"command":0,"value":0.0,"isResponse":false,"stateData":"LED switching OFF","name":"LedActuator","timeStamp":"2022-04-15T03:51:53.645666900Z","statusCode":0,"typeID":2001,"locationID":"ConstrainedDevice","latitude":0.0,"longitude":0.0,"elevation":0.0,"timeStampMillis":1649994713645}
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.connection.MqttClientConnector deliveryComplete
INFO: Delivery complete: 
	Msg ID:           0
	Session Present:  false
	Is Complete:      true
	Topics:           PIOT/ConstrainedDevice/ActuatorCmd,
Apr 14, 2022 11:51:53 PM programmingtheiot.gda.connection.MqttClientConnector deliveryComplete
INFO: Delivered MQTT message with ID: 0
.
.
.
Apr 14, 2022 11:52:32 PM programmingtheiot.gda.app.DeviceDataManager stopManager
INFO: Successfully disconnected MQTT client from broker.
Apr 14, 2022 11:52:32 PM programmingtheiot.part04.integration.connection.CloudClientConnectorTest testIntegratedCloudClientConnectAndDisconnect
INFO: Test complete.