Bug fixed on meep-iot with multiple registrations (755db7ee) · Commits · MEC - Multi-access Edge Computing / ETSI MEC Sandbox

go-packages/meep-iot-mgr/iot-mgr.go

+5 −5

Changes for go-packages/meep-iot-mgr/iot-mgr.go: 5 added lines, 5 removed lines.
Original line number	Diff line number	Diff line
		@@ -313,11 +313,6 @@ func (tm *IotMgr) RegisterIotPlatformInfo(iotPlatformInfo IotPlatformInfo) (err
		if iotPlatformInfo.Enabled {
		iotPlatformInfo.oneM2M = nil
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo: ", iotPlatformInfo.CustomServicesTransportInfo)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint.Addresses: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Addresses)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint.Uris: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Uris)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Protocol: ", iotPlatformInfo.CustomServicesTransportInfo[0].Protocol)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Name: ", iotPlatformInfo.CustomServicesTransportInfo[0].Name)
		if len(iotPlatformInfo.CustomServicesTransportInfo) == 0 \|\|
		iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint == nil \|\|
		(len(iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Addresses) == 0 &&
		@@ -325,6 +320,11 @@ func (tm *IotMgr) RegisterIotPlatformInfo(iotPlatformInfo IotPlatformInfo) (err
		len(iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Fqdn) == 0) {
		log.Error("RegisterIotPlatformInfo: Cannot use provided CustomServicesTransportInfo")
		} else {
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint.Addresses: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Addresses)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Endpoint.Uris: ", iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Uris)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Protocol: ", iotPlatformInfo.CustomServicesTransportInfo[0].Protocol)
		log.Debug("RegisterIotPlatformInfo: CustomServicesTransportInfo[0].Name: ", iotPlatformInfo.CustomServicesTransportInfo[0].Name)
		var pltf *sssmgr.SssMgr = nil
		var err error
		if len(iotPlatformInfo.CustomServicesTransportInfo[0].Endpoint.Addresses) != 0 {

go-packages/meep-sss-mgr/http.go

+44 −28

Changes for go-packages/meep-sss-mgr/http.go: 44 added lines, 28 removed lines.
Original line number	Diff line number	Diff line
		@@ -13,6 +13,7 @@ import (
		"reflect"
		"strconv"
		"strings"
		"sync"
		"time"

		log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger"
		@@ -21,6 +22,7 @@ import (

		type SssMgrHttp struct {
		notify func(map[string]interface{})
		path string
		}

		const (
		@@ -30,7 +32,9 @@ const (
		)

		var (
		notificationServerStarted bool = false
		notificationMu sync.Mutex
		notificationServerStarted bool
		notificationByPath = map[string]func(map[string]interface{}){}
		)

		func NewSssMgrHttp() (http_mgr *SssMgrHttp) {
		@@ -42,33 +46,50 @@ func (http_mgr SssMgrHttp) init(tm SssMgr, notify func(map[string]interface{})
		log.Info(">>> http.init: cse_name: ", tm.cse_name, " host_id: ", tm.hostId, " port: ", tm.port)

		http_mgr.notify = notify
		http_mgr.path = "/" + tm.cse_name

		go func() {
		if notificationServerStarted {
		return
		}
		notificationMu.Lock()
		notificationByPath[http_mgr.path] = notify
		startServer := !notificationServerStarted
		notificationServerStarted = true
		log.Info("Init: Starting OneM2M Notification server on /" + tm.cse_name + ", port: " + notificationPort)
		http.HandleFunc("/"+tm.cse_name, http_mgr.handleRoot)
		err := http.ListenAndServe(":"+notificationPort, nil)
		notificationMu.Unlock()

		if !startServer {
		log.Info("Init: Notification server already running, registered ", http_mgr.path)
		return nil
		}

		mux := http.NewServeMux()
		mux.HandleFunc("/", handleNotification)
		go func() {
		log.Info("Init: Starting OneM2M Notification server, port: " + notificationPort)
		err := http.ListenAndServe(":"+notificationPort, mux)
		if err != nil {
		log.Error(err.Error())
		notificationMu.Lock()
		notificationServerStarted = false
		notificationMu.Unlock()
		return
		}

		log.Info("<<< Init: Terminating OneM2M Notification server")
		notificationMu.Lock()
		notificationServerStarted = false
		notificationMu.Unlock()
		}()

		return nil
		}

		func (http_mgr SssMgrHttp) handleRoot(w http.ResponseWriter, r http.Request) {
		func handleNotification(w http.ResponseWriter, r *http.Request) {
		log.Debug(">>> handleRoot: ", r)

		w.Header().Set("Content-Type", "application/json; charset=UTF-8")

		notificationMu.Lock()
		notify := notificationByPath[r.URL.Path]
		notificationMu.Unlock()

		// Get URL
		u, _ := url.Parse(r.URL.String())
		log.Info("url: ", u.RequestURI())
		@@ -96,8 +117,8 @@ func (http_mgr SssMgrHttp) handleRoot(w http.ResponseWriter, r http.Request) {
		// Call callback
		m := body["m2m:sgn"]
		log.Info("handleRoot: m: ", m)
		if http_mgr.notify != nil {
		http_mgr.notify(m)
		if notify != nil {
		notify(m)
		}
		} else {
		log.Warn("handleRoot: Unsupported method", r.Method)
		@@ -111,6 +132,10 @@ func (http_mgr SssMgrHttp) handleRoot(w http.ResponseWriter, r http.Request) {
		func (http_mgr *SssMgrHttp) uninit() (err error) {
		log.Info(">>> uninit")

		notificationMu.Lock()
		delete(notificationByPath, http_mgr.path)
		notificationMu.Unlock()

		return nil
		}

		@@ -138,23 +163,15 @@ func (http_mgr *SssMgrHttp) send(p_ctx SssMgrBindingProtocolContext) (err error,
		if to != p_ctx.name && !strings.Contains(to, "/") && p_ctx.name != "" {
		to = p_ctx.name + "/" + to
		}
		scheme := p_ctx.scheme
		// if p_ctx.scheme == "https" {
		// scheme = "https://"
		// }
		if p_ctx.path != "" && scheme != "" && p_ctx.host != "" && p_ctx.port != 0 {
		if strings.HasPrefix(p_ctx.path, "/") {
		p_ctx.path = p_ctx.path[1:len(p_ctx.path)]
		}
		url = scheme + "://" + p_ctx.host + ":" + strconv.Itoa(p_ctx.port) + "/" + p_ctx.path + "/" + to
		} else if scheme != "" && p_ctx.host != "" && p_ctx.port != 0 {
		url = scheme + "://" + p_ctx.host + ":" + strconv.Itoa(p_ctx.port) + "/" + to
		} else if p_ctx.host != "" && p_ctx.port != 0 {
		url = p_ctx.host + ":" + strconv.Itoa(p_ctx.port) + "/" + to
		scheme := "https://"
		if p_ctx.port != 443 {
		scheme = "http://"
		}
		basePath := strings.Trim(p_ctx.path, "/")
		if basePath != "" {
		url = scheme + p_ctx.host + ":" + strconv.Itoa(p_ctx.port) + "/" + basePath + "/" + to
		} else {
		err := errors.New("Invalid p_ctx")
		log.Error("send: ", err.Error())
		return err, nil
		url = scheme + p_ctx.host + ":" + strconv.Itoa(p_ctx.port) + "/" + to
		}
		// Set the method
		method := ""
		@@ -212,7 +229,6 @@ func sendRequest(method string, url string, headers http.Header, body io.Reader,
		log.Debug(">>> sendRequest: url: ", url)
		log.Debug(">>> sendRequest: headers: ", headers)

		//
		var bodyBytes []byte
		var err error
		if body != nil {

go-packages/meep-sss-mgr/mqtt.go

+55 −31

Changes for go-packages/meep-sss-mgr/mqtt.go: 55 added lines, 31 removed lines.
Original line number	Diff line number	Diff line
		@@ -18,13 +18,12 @@ type SssMgrMqtt struct {
		opts *mqtt.ClientOptions
		client mqtt.Client
		sent_topic string
		notify func(map[string]interface{})
		respMu sync.Mutex
		pending map[uint16]chan mqtt.Message
		responses map[uint16]mqtt.Message
		}

		var _notify func(map[string]interface{}) = nil

		var _sync_response *sync.WaitGroup = nil // Used to synchronize the response
		var _responses map[uint16]mqtt.Message

		func onMessageReceived(client mqtt.Client, msg mqtt.Message) {
		go func() {
		log.Info("onMessageReceived: Received message: ", msg.Payload(), " on topic ", msg.Topic())
		@@ -47,23 +46,36 @@ func onMessageReceivedReq(client mqtt.Client, msg mqtt.Message) {
		}()
		}

		func onMessageReceivedResp(client mqtt.Client, msg mqtt.Message) {
		func (broker_mqtt *SssMgrMqtt) onMessageReceivedResp(client mqtt.Client, msg mqtt.Message) {
		go func() {
		log.Info("onMessageReceivedResp: Received message: ", string(msg.Payload()), " on topic ", msg.Topic())
		defer _sync_response.Done()
		_responses[msg.MessageID()] = msg
		broker_mqtt.respMu.Lock()
		defer broker_mqtt.respMu.Unlock()
		if ch, ok := broker_mqtt.pending[msg.MessageID()]; ok {
		delete(broker_mqtt.pending, msg.MessageID())
		ch <- msg
		return
		}
		if broker_mqtt.responses == nil {
		return
		}
		broker_mqtt.responses[msg.MessageID()] = msg
		}()
		}

		func onMessageReceivedNot(client mqtt.Client, msg mqtt.Message) {
		func (broker_mqtt *SssMgrMqtt) onMessageReceivedNot(client mqtt.Client, msg mqtt.Message) {
		go func() {
		log.Info("onMessageReceivedNot: Received message: ", string(msg.Payload()), " on topic ", msg.Topic())
		//{\"fr\": \"/laboai-acme-ic-cse\", \"to\": \"mqtt://172.29.10.56:1883\", \"ot\": \"20250327T134413,103223\", \"op\": 5, \"rqi\": \"6400213617020980613\", \"rvi\": \"4\", \"drt\": 1, \"pc\": {\"m2m:sgn\": {\"nev\": {\"net\": 3, \"rep\": {\"m2m:cin\": {\"cnf\": \"text/plain:0\", \"con\": \"toto\", \"rn\": \"1234\", \"ri\": \"cin3503344383965775257\", \"pi\": \"cnt1992620086833081496\", \"ct\": \"20250327T134413,093784\", \"lt\": \"20250327T134413,093784\", \"ty\": 4, \"cs\": 4, \"st\": 2, \"et\": \"20300326T134347,108295\"}}}, \"sur\": \"/laboai-acme-ic-cse/sub3169944238694067800\"}}} on topic /oneM2M/req/laboai-acme-ic-cse/mqtt:::172.29.10.56:1883/json"
		// Prepare response
		s := strings.Split(msg.Topic(), "/") // /oneM2M/req/laboai-acme-ic-cse/mqtt:::172.29.10.56:1883/json
		if len(s) < 6 {
		log.Warn("onMessageReceivedNot: unexpected topic: ", msg.Topic())
		return
		}
		resp_topic := "/oneM2M/resp/" + s[4] + "/" + s[3] + "/" + s[5]
		log.Debug("onMessageReceivedNot: resp_topic: ", resp_topic)
		if _notify != nil {
		if broker_mqtt.notify != nil {
		var d map[string]interface{}
		err := json.Unmarshal(msg.Payload(), &d)
		if err != nil {
		@@ -88,7 +100,7 @@ func onMessageReceivedNot(client mqtt.Client, msg mqtt.Message) {
		}
		m := body["m2m:sgn"].(map[string]interface{})
		log.Info("handleRoot: m: ", m)
		_notify(m)
		broker_mqtt.notify(m)
		} else {
		log.Warn("onMessageReceivedNot: pc entry not found")
		return
		@@ -117,12 +129,13 @@ func (broker_mqtt SssMgrMqtt) init(tm SssMgr, notify func(map[string]interface
		log.Info(">>> init")

		broker_mqtt.running = false

		_notify = notify
		broker_mqtt.notify = notify
		broker_mqtt.pending = make(map[uint16]chan mqtt.Message)
		broker_mqtt.responses = make(map[uint16]mqtt.Message)

		broker_mqtt.opts = mqtt.NewClientOptions()
		broker_mqtt.opts.SetDefaultPublishHandler(onMessageReceived)
		broker_mqtt.opts.SetClientID("AdvantEDGE.meep-sss-mgr")
		broker_mqtt.opts.SetClientID(fmt.Sprintf("meep-iot-%s-%d", tm.hostId, tm.port))
		broker_mqtt.opts.OnConnect = connectHandler
		broker_mqtt.opts.OnConnectionLost = connectLostHandler
		broker_mqtt.opts.CleanSession = true
		@@ -152,7 +165,7 @@ func (broker_mqtt SssMgrMqtt) init(tm SssMgr, notify func(map[string]interface
		// token.Wait()
		tp := "/oneM2M/resp/+/" + tm.cse_name + "/#"
		log.Info("init: Subscribe to: ", tp)
		token = broker_mqtt.client.Subscribe(tp, 0, onMessageReceivedResp) // qos:0
		token = broker_mqtt.client.Subscribe(tp, 0, broker_mqtt.onMessageReceivedResp) // qos:0
		if token.Error() != nil {
		broker_mqtt.client.Disconnect(250)
		log.Error(token.Error())
		@@ -161,7 +174,7 @@ func (broker_mqtt SssMgrMqtt) init(tm SssMgr, notify func(map[string]interface
		token.Wait()
		tp = "/oneM2M/req/" + tm.cse_name + "/+/#"
		log.Info("init: Subscribe to: ", tp)
		token = broker_mqtt.client.Subscribe(tp, 1, onMessageReceivedNot) // qos:1
		token = broker_mqtt.client.Subscribe(tp, 1, broker_mqtt.onMessageReceivedNot) // qos:1
		if token.Error() != nil {
		broker_mqtt.client.Disconnect(250)
		log.Error(token.Error())
		@@ -169,9 +182,6 @@ func (broker_mqtt SssMgrMqtt) init(tm SssMgr, notify func(map[string]interface
		}
		token.Wait()

		_sync_response = new(sync.WaitGroup) // Used to synchronize the response
		_responses = make(map[uint16]mqtt.Message, 0)

		broker_mqtt.running = true
		log.Info("init: Client is connected")

		@@ -196,8 +206,11 @@ func (broker_mqtt *SssMgrMqtt) uninit() (err error) {
		broker_mqtt.client.Disconnect(250)
		broker_mqtt.running = false

		_sync_response = nil
		_responses = nil
		broker_mqtt.respMu.Lock()
		broker_mqtt.pending = nil
		broker_mqtt.responses = nil
		broker_mqtt.notify = nil
		broker_mqtt.respMu.Unlock()

		return nil
		}
		@@ -265,13 +278,29 @@ func (broker_mqtt *SssMgrMqtt) send(p_ctx SssMgrBindingProtocolContext) (err err
		token := broker_mqtt.client.Publish(broker_mqtt.sent_topic, 0, false, string(content))
		token.Wait()

		// Syncronization with the response
		// Synchronization with the response. Match on this client's publish id so a
		// second platform cannot consume or panic on the first client's WaitGroup.
		messageID := token.(*mqtt.PublishToken).MessageID()
		broker_mqtt.respMu.Lock()
		if val, ok := broker_mqtt.responses[messageID]; ok {
		delete(broker_mqtt.responses, messageID)
		broker_mqtt.respMu.Unlock()
		return decodeMqttResponse(val)
		}
		ch := make(chan mqtt.Message, 1)
		if broker_mqtt.pending == nil {
		broker_mqtt.pending = make(map[uint16]chan mqtt.Message)
		}
		broker_mqtt.pending[messageID] = ch
		broker_mqtt.respMu.Unlock()

		log.Debug("send: Start waiting for the response")
		_sync_response.Add(1)
		_sync_response.Wait()
		val := <-ch
		log.Debug("send: Got the response")
		if val, ok := _responses[token.(*mqtt.PublishToken).MessageID()]; ok {
		delete(_responses, token.(*mqtt.PublishToken).MessageID())
		return decodeMqttResponse(val)
		}

		func decodeMqttResponse(val mqtt.Message) (err error, resp interface{}) {
		log.Debug("send: Get the response: ", string(val.Payload()))
		var d map[string]interface{}
		err = json.Unmarshal(val.Payload(), &d)
		@@ -284,10 +313,5 @@ func (broker_mqtt *SssMgrMqtt) send(p_ctx SssMgrBindingProtocolContext) (err err
		log.Debug("send: TypeOf(r): ", reflect.TypeOf(r))
		return nil, r
		}
		return err, nil
		} else {
		log.Info("send: No response for messageID: ", token.(*mqtt.PublishToken).MessageID())
		}

		return nil, nil
		}

go-packages/meep-sss-mgr/onem2m-mgr.go

+193 −83

File changed.

Preview size limit exceeded, changes collapsed.

go-packages/meep-sss-mgr/onem2m-mgr_test.go

+31 −4

Changes for go-packages/meep-sss-mgr/onem2m-mgr_test.go: 31 added lines, 4 removed lines.
Original line number	Diff line number	Diff line
		@@ -281,7 +281,7 @@ func TestOneM2M_createAE_FLEXHttp(t *testing.T) {
		SensorPosition: nil,
		IotPlatformId: "0a692154-0f93-11f0-b554-df2f6a756f6a",
		}
		new_sensor_ae, err := tm.OneM2M_create(sensor_ae, "")
		new_sensor_ae, err := tm.OneM2M_create(sensor_ae, "", "CMyAE")
		if err != nil {
		t.Fatalf("Failed to create new AE sensor")
		}
		@@ -304,7 +304,7 @@ func TestOneM2M_createAE_FLEXHttp(t *testing.T) {
		sensor_flex.Flex["green"] = 20
		sensor_flex.Flex["blue"] = 20
		sensorPath := new_sensor_ae.SensorIdentifier
		new_sensor_flex, err := tm.OneM2M_create(sensor_flex, sensorPath)
		new_sensor_flex, err := tm.OneM2M_create(sensor_flex, sensorPath, "CMyAE")
		if err != nil {
		t.Fatalf("Failed to create new CNT sensor")
		}
		@@ -321,11 +321,11 @@ func TestOneM2M_createAE_FLEXHttp(t *testing.T) {
		fmt.Println("len=", len(sensors))
		fmt.Println("sensors", sensors)

		err = tm.OneM2M_delete(new_sensor_flex)
		err = tm.OneM2M_delete(new_sensor_flex, "CMyAE")
		if err != nil {
		t.Fatalf("Failed to create new sensor")
		}
		err = tm.OneM2M_delete(new_sensor_ae)
		err = tm.OneM2M_delete(new_sensor_ae, "CMyAE")
		if err != nil {
		t.Fatalf("Failed to create new sensor")
		}
		@@ -1361,3 +1361,30 @@ func dataNotify(map[string]interface{}) {
		log.Debug(">>> dataNotify")

		}

		func TestSecondIotPlatformKeepsFirstRegistration(t *testing.T) {
		log.MeepTextLogInit(t.Name())

		first, err := NewSssMgr(tmName, tmNamespace, "REST_HTTP", "127.0.0.1", 18080, "platform-1", "cse-1", nil, nil, nil)
		if err != nil {
		t.Fatal(err)
		}
		second, err := NewSssMgr(tmName, tmNamespace, "REST_HTTP", "127.0.0.1", 18081, "platform-2", "cse-2", nil, nil, nil)
		if err != nil {
		t.Fatal(err)
		}
		defer first.DeleteSssMgr()
		defer second.DeleteSssMgr()

		platform1, ok1 := platformById("platform-1")
		platform2, ok2 := platformById("platform-2")
		if !ok1 \|\| !ok2 {
		t.Fatalf("both platforms should stay registered, got %v %v", ok1, ok2)
		}
		if platform1.Name != "cse-1" \|\| platform1.Port != 18080 {
		t.Fatalf("first platform was overwritten: %+v", platform1)
		}
		if platform2.Name != "cse-2" \|\| platform2.Port != 18081 {
		t.Fatalf("second platform was not stored: %+v", platform2)
		}
		}