Commit 6c78a3f8 authored by Muhammad Umair Khan's avatar Muhammad Umair Khan
Browse files

Fixes #15: Update MQTT subscription scope and implement dynamic V2X message parsing 

This commit addresses the restrictive MQTT subscription and hardcoded notification logic in the VIS Traffic Manager.

Changes introduced:
- Upgraded the MQTT subscription wildcard from '+' to '#' to properly capture deeply nested hierarchical V2X sub-topics.
- Removed the hardcoded '3gpp/v2x/obu/vam' topic check in the message receiver.
- Implemented a dynamic topic parser that strictly adheres to the ETSI GS MEC 030 specification (Section 6.5.14). The parser now accurately extracts the 'stdOrganization', 'msgType', and 'msgProtocolVersion' directly from the MQTT topic suffix.
- Created a comprehensive lookup map utilizing the exact ETSI TS 102 894-2 integer constants (e.g., CAM=2, DENM=1, VAM=16), enabling native support for all 20 standardized V2X message types instead of relying on a hardcoded VAM value.
- Implemented robust fallback mechanisms to safely apply default values if publishers utilize abbreviated legacy topic structures.
parent 355584ff

go-packages/meep-vis-traffic-mgr/mqtt.go

+55 −4
Original line number Diff line number Diff line
@@ -21,6 +21,7 @@ import ( 
	"encoding/hex"

	"errors"

	"strings"

	"strconv"



	log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"

	mqtt "github.com/eclipse/paho.mqtt.golang"
@@ -34,14 +35,63 @@ type message_broker_mqtt struct { 
}



var _v2x_notify func(v2xMessage []byte, v2xType int32, msgProtocolVersion int32, stdOrganization string, longitude *float32, latitude *float32)

var _tm_topic string



var msgTypeMap = map[string]int32{

	"denm":              1,

	"cam":               2,

	"poi":               3,

	"spatem":            4,

	"mapem":             5,

	"ivim":              6,

	"ev_rsr":            7,

	"tistpgtransaction": 8,

	"srem":              9,

	"ssem":              10,

	"evcsn":             11,

	"saem":              12,

	"rtcmem":            13,

	"cpm":               14,

	"imzm":              15,

	"vam":               16,

	"dsm":               17,

	"mim":               18,

	"mvm":               19,

	"mcm":               20,

}



func onMessageReceived(client mqtt.Client, msg mqtt.Message) {

	go func() {

		log.Info("onMessageReceived: Received message: ", msg.Payload(), " on topic ", msg.Topic())

		if _v2x_notify != nil {

			if msg.Topic() == "3gpp/v2x/obu/vam" { // FIXME FSCOM Need to manage how to extract message type & message version

				_v2x_notify(msg.Payload(), 16, 2, "ETSI", nil, nil)

			var stdOrg string = "ETSI" // default

			var msgType int32 = 16       // default VAM

			var version int32 = 2        // default



			topicSuffix := strings.TrimPrefix(msg.Topic(), _tm_topic+"/")

			parts := strings.Split(topicSuffix, "/")



			if len(parts) >= 1 {

				mTypeStr := parts[0]

				if len(parts) == 3 {

					stdOrg = parts[0]

					mTypeStr = parts[1]

					if v, err := strconv.ParseInt(parts[2], 10, 32); err == nil {

						version = int32(v)

					}

				} else if len(parts) == 2 {

					mTypeStr = parts[0]

					if v, err := strconv.ParseInt(parts[1], 10, 32); err == nil {

						version = int32(v)

					}

				}

				

				if val, ok := msgTypeMap[strings.ToLower(mTypeStr)]; ok {

					msgType = val

				}

			}



			_v2x_notify(msg.Payload(), msgType, version, stdOrg, nil, nil)

		} else {

			log.Info("onMessageReceived: null pointer for the callback")

		}
@@ -86,14 +136,15 @@ func (broker_mqtt *message_broker_mqtt) Init(tm *TrafficMgr) (err error) { 
	token.Wait()



	// Subscribe

	log.Info("Init: Subscribe to: ", tm.topic+"/+")

	token = broker_mqtt.client.Subscribe(tm.topic+"/+", 0, nil) // qos:0

	log.Info("Init: Subscribe to: ", tm.topic+"/#")

	token = broker_mqtt.client.Subscribe(tm.topic+"/#", 0, nil) // qos:0

	if token.Error() != nil {

		log.Error(token.Error())

		return token.Error()

	}

	token.Wait()



	_tm_topic = tm.topic

	_v2x_notify = broker_mqtt.v2x_notify

	if _v2x_notify == nil {

		log.Error("Init: _v2x_notify is nil")