How to differentiate the data of one topic or another coming from Mercure hub if the client is suscribed to two topics

Question

I have a a dashboard on a Symony app with two ChartJS graphs, one for temperature data, and the other for pressure data. I need update both on realtime; for that purpose I try to use MercureBundle with two topics: ['realtime-notif/temperature/{sensorId}', 'realtime-notif/pressure/{sensorId}']. Although the topics sound similar, the logic to conform datas are differents because the two ChartJS are differents, and for that I have two Messenger messages handlers with AMQP queue, one publish a mercure update in topic 'realtime-notif/temperature/{sensorId}' and the other message handler class publish in 'realtime-notif/pressure/{sensorId}'. I will try to summarize the code to be concise.

#mercure.yaml:
mercure:
    hubs:
        default:
            url: '%env(MERCURE_URL)%'
            public_url: '%env(MERCURE_PUBLIC_URL)%'
            jwt:
                secret: '%env(MERCURE_JWT_SECRET)%'
                publish: ['realtime-notif/temperature/{sensorId}', 'realtime-notif/pressure/{sensorId}']
                subscribe: ['realtime-notif/temperature/{sensorId}', 'realtime-notif/pressure/{sensorId}']

#The TemperatureMessageHandler class: 
class TemperatureMessageHandler implements MessageHandlerInterface
{

    private $mercureHub;
    private $managerRegistry;

    public function __construct(HubInterface $mercureHub, ManagerRegistry $managerRegistry)
    {
        $this->mercureHub = $mercureHub;
        $this->managerRegistry = managerRegistry;
    }

    public function __invoke(TemperatureMessage $message)
    {
        try {
            $graphId=$message->getGraphId();
            $lastElapsedTime=$message->getLastElapsedTime();
            
            $em=$this->managerRegistry->getManager();

            $storedData = $em->getRepository(Temperature::class)->findLastRecordsForGraph($graphId, $lastElapsedTime);
            
            /**
             Set the data source for the temperature graph to a specific format from $toredData
            **/
            $formatedChartData = [];
                **....**

            $update = new Update(
                    sprintf('realtime-notif/temperature/%s', $graphId),
                    \json_encode($$formatedChartData),
                    true
            );

            $this->mercureHub->publish($update);
        } catch (\Exception $exc) {
            
        }
    }
}

And,

#The PressureMessageHandler class:
 class PressureMessageHandler implements MessageHandlerInterface
{

    private $mercureHub;
    private $managerRegistry;

    public function __construct(HubInterface $mercureHub, ManagerRegistry $managerRegistry)
    {
        $this->mercureHub = $mercureHub;
        $this->managerRegistry = managerRegistry;
    }

    public function __invoke(PressureMessage $message)
    {
        try {
            $graphId = $message->getGraphId();
            $lastElapsedTime = $message->getLastElapsedTime();
            
            $em = $this->managerRegistry->getManager();

            $storedData = $em->getRepository(Pressure::class)->findLastRecordsForGraph($graphId, $lastElapsedTime);
            
            /**
             Set the data source for the pressure graph to a specific format from $toredData
            **/
            $formatedChartData = [];
                **....**
            

            $update = new Update(
                    sprintf('realtime-notif/pressure/%s', $graphId),
                    \json_encode($$formatedChartData),
                    true
            );

            $this->mercureHub->publish($update);
        } catch (\Exception $exc) {
            
        }
    }
}

The problem for me is I don't know how to differentiate on the client side if the data received from the Mercure hub is from temperature topic or pressure topic in the message event of EventSource object.

<script type="text/javascript">
                $(document).ready(function () {
                    /**Create two graph on page ready **/
                    let temperatureGraphObject = createTemperatureGraph(canvasTemperaturaGraph);
                    let pressureGRaphObject = createPressureGraph(canvasPressureGraph);
                    
                    /** 
                    I have two function updateTemperatureGraph(temperatureGraphObject, newTemperaturaData) and updatePressureGraph(pressureGraphObject, newPresureData)
                    **/
                    
                    /**Subscribe client to topics for data updates **/
                    {% set topics = ['realtime-notif/temperature/'~temperatureSensorId, 'realtime-notif/pressure/'~pressureSensorId] %}

                    const eventSource = new EventSource("{{ mercure(topics, { subscribe:topics})|escape('js')}}", {withCredentials: true});

                    eventSource.onopen = function () {
                        console.log('New socket connection!');
                    };

                    eventSource.onmessage = function (e) {
                        console.log('New data received');
                        var data = JSON.parse(e.data);
                        
                        /** 
                        The problem is here, how differentiate the topics data to call updateTemperaturaGraph(temperatureGraphObject, data) or  updatePressureGraph(pressureGraphObject, data)
                        **/                     
                    };

                    eventSource.onerror = function () {
                        console.log('Socket connection lost!');
                    };
                });
            </script>

So, how differentiate the topics data to call updateTemperaturaGraph(temperatureGraphObject, data) or updatePressureGraph(pressureGraphObject, data) into onmessage event?

If I subscribe the client to only one topic all data received will be of kind of the topic graph, and of course the graph is updated correctly.

Answer 1

The solution is set de type attribute for Update class constructor that you want to publish in your Mercure hub from the MessageHandler related class. So, for the messages related to temperature notification we will set the type attribute = 'temperatureUpdate', and for the message related to pressure notification we will set the type attribute = 'pressureUpdate'.

The __invoke function on TemperatureMessageHandler:

public function __invoke(TemperatureMessage $message)
{
    try {
        $graphId = $message->getGraphId();
        
        // Collect the data for the update
        $data = [/* ... */];

        $update = new Update(
            sprintf('realtime-notif/temperature/%s', $graphId),
            \json_encode($data),
            true,
            null,
            'temperatureUpdate'
        );

        $this->mercureHub->publish($update);
    } catch (\Exception $exc) {
        
    }
}

The __invoke function on PressureMessageHandler:

public function __invoke(PressureMessage $message)
{
    try {
        $graphId = $message->getGraphId();

        // Collect the data for the update
        $data = [/* ... */];

        $update = new Update(
            sprintf('realtime-notif/pressure/%s', $graphId),
            \json_encode($data),
            true,
            null,
            'pressureUpdate'
        );

        $this->mercureHub->publish($update);
    } catch (\Exception $exc) {
        
    }
}

On the client side, is mandatory to create two new EventListeners for the EventSource object, with the name equals to the new types created. Every one new listener will treat the related messages types published in the Mercure hub:

<script type="text/javascript">
    $(document).ready(function () {
        /**Create two graph on page ready **/
        let temperatureGraphObject = createTemperatureGraph(canvasTemperaturaGraph);
        let pressureGRaphObject = createPressureGraph(canvasPressureGraph);
        
        /** 
        I have two function updateTemperatureGraph(temperatureGraphObject, newTemperaturaData) and updatePressureGraph(pressureGraphObject, newPresureData)
        **/
        
        /**Subscribe client to topics for data updates **/
        {% set topics = ['realtime-notif/temperature/'~temperatureSensorId, 'realtime-notif/pressure/'~pressureSensorId] %}

        const eventSource = new EventSource("{{ mercure(topics, { subscribe:topics})|escape('js')}}", {withCredentials: true});

        eventSource.onopen = function () {
            console.log('New socket connection!');
        };

        eventSource.addEventListener("temperaturaUpdate", function (e) {
            let parsedData = null;
            try {
                parsedData = JSON.parse(e.data);
            } catch (error) {
                console.log(error);
            }
            
            if (parsedData) {
                updateTemperatureGraph(temperatureGraphObject, parsedData);
            }
        }, false);
        
        eventSource.addEventListener("pressureUpdate", function (e) {
            let parsedData = null;
            try {
                parsedData = JSON.parse(e.data);
            } catch (error) {
                console.log(error);
            }
            
            if (parsedData) {
                updatePressureGraph(pressureGraphObject, parsedData);
            }
        }, false);

        eventSource.onerror = function () {
            console.log('Socket connection lost!');
        };
    });
</script>

In this way, the Mercure hub publishes the classified messages, and each EventListener will be in charge of processing the corresponding message in the order in which it arrives at the subscribed client, regardless of the topics to which it is subscribed.