@ RobotKeyword ( "Subscribe to MQTT Broker and validate that it received a specific message" ) @ ArgumentNames ( { "broker" , "clientId" , "topic" , "expectedPayload" , "timeout" } ) public void subscribeToMQTTAndValidate ( String broker , String clientId , String topic , String expectedPayload , long timeout ) { MqttClient client = null ; try { MqttClientPersistence persistence = new MemoryPersistence ( ) ; client = new MqttClient ( broker , clientId , persistence ) ; // set clean session to false so the state is remembered across // sessions MqttConnectOptions connOpts = new MqttConnectOptions ( ) ; connOpts . setCleanSession ( false ) ; // set callback before connecting so prior messages are delivered as // soon as we connect MQTTResponseHandler handler = new MQTTResponseHandler ( ) ; client . setCallback ( handler ) ; System . out . println ( "*INFO:" + System . currentTimeMillis ( ) + "* Connecting to broker: " + broker ) ; client . connect ( connOpts ) ; System . out . println ( "*INFO:" + System . currentTimeMillis ( ) + "* Subscribing to topic: " + topic ) ; client . subscribe ( topic ) ; System . out . println ( "*INFO:" + System . currentTimeMillis ( ) + "* Subscribed to topic: " + topic ) ; // now loop until either we receive the message in the topic or // timeout System . out . println ( "*INFO:" + System . currentTimeMillis ( ) + "* Waiting for message to arrive" ) ; boolean validated = false ; byte [ ] payload ; MqttMessage message ; long endTime = System . currentTimeMillis ( ) + timeout ; while ( true ) { /* * If expected payload is empty, all we need to validate is * receiving the message in the topic. If expected payload is * not empty, then we need to validate that it is contained in * the actual payload */ message = handler . getNextMessage ( timeout ) ; if ( message != null ) { // received a message in the topic payload = message . getPayload ( ) ; String payloadStr = new String ( payload ) ; if ( expectedPayload . isEmpty ( ) || ( payloadStr . matches ( expectedPayload ) ) ) { validated = true ; break ; } } // update timeout to remaining time and check if ( ( timeout = endTime - System . currentTimeMillis ( ) ) <= 0 ) { System . out . println ( "*DEBUG:" + System . currentTimeMillis ( ) + "* timeout: " + timeout ) ; break ; } } if ( ! validated ) { throw new RuntimeException ( "The expected payload didn't arrive in the topic" ) ; } } catch ( MqttException e ) { throw new RuntimeException ( e . getLocalizedMessage ( ) ) ; } finally { try { client . disconnect ( ) ; } catch ( MqttException e ) { // empty } } }
Subscribe to an MQTT broker and validate that a message with specified payload is received