PIOT-CDA-10-003: Update MqttClientConnector to subscribe to ActuatorData command messages from the GDA

[labbenchstudios]

Description

• Update MqttClientConnector to subscribe to ActuatorData command messages from the GDA and send received messages to the IDataMessageListener instance.

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

• Within the MqttClientConnector class:
  • Add the following import statement:

    from programmingtheiot.data.DataUtil import DataUtil

• Within the MqttClientConnector class constructor, set a class-scoped variable named self.dataMsgListener = None
  • Add the following function to set the IDataMessageListener reference:

    def setDataMessageListener(self, listener: IDataMessageListener = None):
    if listener:
    self.dataMsgListener = listener

• Make sure the above method is invoked from within DeviceDataManager, AFTER creating the MqttClientConnector instance, be sure to call the above method passing self as the listener reference:

  # TODO: Call this from within `DeviceDataManager's` constructor
  if self.enableMqttClient:
  self.mqttClient = MqttClientConnector()
  self.mqttClient.setDataMessageListener(self)

• Within MqttClientConnector, create a new callback method to handle incoming subscription messages for actuator command messages, as follows:

  def onActuatorCommandMessage(self, client, userdata, msg):
  logging.info('[Callback] Actuator command message received. Topic: %s.', msg.topic)
  
  if self.dataMsgListener:
      try:
          # assumes all data is encoded using UTF-8 (between GDA and CDA)
          actuatorData = DataUtil().jsonToActuatorData(msg.payload.decode('utf-8'))
  
          self.dataMsgListener.handleActuatorCommandMessage(actuatorData)
      except:
          logging.exception("Failed to convert incoming actuation command payload to ActuatorData: ")

• Add actuator command subscriptions to the onConnect() callback function.

  • NOTE: This will be invoked once the connectClient() is called and the connection to the broker succeeds, ensuring the necessary topic subscriptions are called after a successful broker connection.
  • IMPORTANT: If you've already added an actuator command topic subscription to your CDA's DeviceDataManager after invoking the connectClient() method on your MQTT client instance within DeviceDataManager (e.g., during the startManager() call), you should remove that logic before moving forward.
  • First, call the generic subscription callback using mqttClient.subscribe(topic, qos). This will register the subscription with the MQTT broker. Note that the topic parameter is the topic name string, and qos is the integer QoS value.
  • Next, add a topic-specific callback using the message_callback_add(sub, callback) function. This will redirect the received subscription messages from mqttClient.subscribe(topic, qos) to a new callback function based on the given topic. This function accepts a sub parameter for the topic name and a callback parameter for the callback function reference.
  • Here's one implementation approach for handling the actuator command topic subscription process described above:

    
    def onConnect(self, client, userdata, flags, rc):
    logging.info('[Callback] Connected to MQTT broker. Result code: ' + str(rc))

  NOTE: Be sure to set self.defaultQos during instantiation!

  self.mqttClient.subscribe( \ topic = ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE.value, qos = self.defaultQos)

  self.mqttClient.message_callback_add( \ sub = ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE.value, \ callback = self.onActuatorCommandMessage)

  
  **Important - how to deal with 'locked up publishing' (e.g., potential deadlock condition)**

• Recall from Lab Module 06, the CDA's MqttClientConnector added the msgInfo.wait_for_publish() line of code after the publish call within the publishMessage method. This was to ensure the message was actually published and received by the broker before returning. However, this CAN cause a deadlock condition, or a never ending 'wait', as the publish call will block all other calls against the underlying MQTT client logic.

  • Why it matters: For Lab Module 10, your CDA's MQTT logic needs to be able to process publish and subscribe events asynchronously so that there's no blocking, or waiting, for a single call to complete before processing incoming (or outgoing) data. Fortunately, it's easy to resolve...
  • How to resolve: It will likely benefit your implementation if you simply comment out the msgInfo.wait_for_publish() call.
    • More detail for those interested: This will avoid a blocking loop within the library that waits until a response is received from the broker, which will also lock out other requests to the underlying infrastructure. For this exercise, it's best to return immediately after the publish request without blocking.
  • How to know if the newly asynchronous publish event succeeded: As you can see in the publish call, an instance of MQTTMessageInfo is returned, which allows you to check - at some later point - if the publish was successful via a call to is_published() (which returns a boolean).
    • There are many ways to check this status, one of which is to trigger a thread that will - asynchronously - invoke the is_published() method on MQTTMessageInfo to see if (or when) it finally completed. This logic can get complicated of course, and while important, is out of scope for this exercise. If you choose to implement this logic, it's up to you how you may want to implement it in an asynchronous manner.
  • Basic implementation approach: As indicated above, simply comment out the msgInfo.wait_for_publish()' code at the end of the method. Your updated implementation ofpublishMessage` may look similar to the following:

    
    def publishMessage(self, resource: ResourceNameEnum = None, msg: str = None, qos: int = ConfigConst.DEFAULT_QOS) -> bool:
    """
    """
    # check validity of resource (topic)
    if not resource:
    logging.warning('No topic specified. Cannot publish message.')
    return False

  check validity of message

  if not msg: logging.warning('No message specified. Cannot publish message to topic: ' + resource.value) return False

  check validity of QoS - set to default if necessary

  if qos < 0 or qos > 2: qos = ConfigConst.DEFAULT_QOS

  publish message, and wait for publish to complete before returning

  msgInfo = self.mqttClient.publish(topic = resource.value, payload = msg, qos = qos)

  The next SLOC is commented out now - recall it was added in Lab Module 06

  msgInfo.wait_for_publish()

  NOTE: The 'True' return no longer guarantees successful publish,

  as it will return before the publish may successfully complete

  return True

Estimate

• Medium

Tests

• Setup

  • Start your local MQTT broker using the default configuration and execute the following - for now, WITHOUT TLS enabled:
  • NOTE: For these tests, be sure to run them against a locally running MQTT broker (e.g., Mosquitto) on your local network. Be sure to run this test without TLS enabled on your MQTT broker - you can enable TLS for later tests.
  • Add a new test
  • In the src/test/python/programmingtheiot/part03/integration/connection package, add a new test case to the existing MqttClientConnectorTest class to test the new subscribe functionality.
  • NOTE: If you already have a test with the same name (e.g., testActuatorCmdPubSub()), just give the new test a new name (such as testNewActuatorCmdPubSub()), and disable all of the other tests in MqttClientConnectorRest using the @unittest.skip("Ignore for now.") annotation).

    
    def testActuatorCmdPubSub(self):
    qos = 1

  NOTE: delay can be anything you'd like - the sleep() calls are simply to slow things down a bit for observation

  delay = self.cfg.getInteger(ConfigConst.MQTT_GATEWAY_SERVICE, ConfigConst.KEEP_ALIVE_KEY, ConfigConst.DEFAULT_KEEP_ALIVE)

  actuatorData = ActuatorData() payload = DataUtil().actuatorDataToJson(actuatorData)

  NOTE: the DefaultDataMessageListener() is just a placeholder for

  handling callbacks from the MQTT client - it is optional

  self.mcc.setDataMessageListener(DefaultDataMessageListener()) self.mcc.connectClient()

  sleep(5)

  self.mcc.publishMessage(resource = ResourceNameEnum.CDA_ACTUATOR_CMD_RESOURCE, msg = payload, qos = qos)

  sleep(delay)

  self.mcc.disconnectClient()

• Run the new test and watch the log output. It should look similar to the following:

  
  2020-11-16 23:23:02,713:MqttClientConnector:INFO:Using requested client ID: CDAMqttClientConnectorTest001
  2020-11-16 23:23:02,713:MqttClientConnector:INFO:   MQTT Broker Host: localhost
  2020-11-16 23:23:02,713:MqttClientConnector:INFO:   MQTT Broker Port: 1883
  2020-11-16 23:23:02,713:MqttClientConnector:INFO:   MQTT Keep Alive:  30
  2020-11-16 23:23:02,714:DataUtil:INFO:Created DataUtil instance.
  2020-11-16 23:23:02,714:DataUtil:DEBUG:Encoding ActuatorData to JSON [pre]  --> name=Not Set,timeStamp=2020-11-16 23:23:02.714167,command=0,hasError=False,statusCode=0,stateData=None,curValue=0.0,actuatorType=0
  2020-11-16 23:23:02,714:DataUtil:INFO:Encoding ActuatorData to JSON [post] --> {
  "timeStamp": "2020-11-16 23:23:02.714167",
  "name": "Not Set",
  "hasError": false,
  "statusCode": 0,
  "isResponse": false,
  "actuatorType": 0,
  "command": 0,
  "stateData": null,
  "curValue": 0.0
  }
  2020-11-16 23:23:02,714:MqttClientConnector:INFO:Attempting to connect to MQTT broker: localhost
  2020-11-16 23:23:02,716:MqttClientConnector:INFO:[Callback] Connected to MQTT broker. Result code: 0
  2020-11-16 23:23:02,716:MqttClientConnector:INFO:[Callback] Subscribed MID: 1
  2020-11-16 23:23:07,719:MqttClientConnector:INFO:[Callback] Actuator command message received. Topic: PIOT/ConstrainedDevice/ActuatorCmd.
  2020-11-16 23:23:37,723:MqttClientConnector:INFO:Disconnecting from MQTT broker: localhost
  2020-11-16 23:23:37,724:MqttClientConnector:INFO:[Callback] Disconnected from MQTT broker. Result code: 0
  ----------------------------------------------------------------------
  Ran 5 tests in 35.014s

OK (skipped=4)

[labbenchstudios]

Clarified potential deadlock condition handling in latest update.