KURA : how to receive MQTT messages using the DataService?

How can I receive the MQTT messages published on a specific topic? I can publish messages on the MQTT broker but I dont know how to receive messages?

Here is my component xml:

<?xml version="1.0" encoding="UTF-8"?>
<scr:component xmlns:scr="http://www.osgi.org/xmlns/scr/v1.1.0" activate="activate" deactivate="deactivate" name="arduinoCommunicator">
   <implementation class="arduinoCommunicator.ArduinoCommunicator"/>
   <reference bind="setDataService" cardinality="1..1" interface="org.eclipse.kura.data.DataService" name="DataService" policy="static" unbind="unsetDataService"/>
   <service>
      <provide interface="org.eclipse.kura.data.DataServiceListener"/>
   </service>
</scr:component>

Here is my bundle activator class, I only kept the needed code, I implement the DataServiceListener, I subscribe to the topic on the onConnectionEstablished() and I imagine that I got the message on the function onMessageArrived, the problem is I don't see the logs that the subscription has been made and the function onMessageArrived() is not executed:

public class ArduinoCommunicator implements DataServiceListener {
public static DataService dataService; 


@Override
public void onConnectionEstablished() {
    String topic="egmkey/device2/cmd/PING";
    System.out.println("connection has been established");
    try {
        ArduinoCommunicator.dataService.subscribe(topic, 1);//egmkey/device2/cmd/PING
        System.out.println("subscription done to topic"+topic);
    } catch (KuraException e) {
        System.out.println("failed to subscribe: "+ e);
    }
}

@Override
    public void onMessageArrived(String topic, byte[] payload, int qos, boolean retained) {
        System.out.println("message received!!!!!!!!!!!!!!!!!!!!!!:"+topic);

    }

}

What have I missed to be able to receive MQTT messages using DataService on KURA? Thanks.

You are not correctly injecting the DataService service into your component. The code below should be closer to what you actually need:

public class ArduinoCommunicator implements DataServiceListener {
    public DataService dataService;

    protected setDataService(DataService dataService) {
        this.dataService = dataService;
    }

    @Override
    public void onConnectionEstablished() {
        String topic = "egmkey/device2/cmd/PING";
        System.out.println("connection has been established");
        try {
            dataService.subscribe(topic, 1);//egmkey/device2/cmd/PING
            System.out.println("subscription done to topic" + topic);
        } catch (KuraException e) {
            System.out.println("failed to subscribe: " + e);
        }
    }

    @Override
    public void onMessageArrived(String topic, byte[] payload, int qos, boolean retained) {
        System.out.println("message received!!!!!!!!!!!!!!!!!!!!!!:" + topic);

    }

}