Commit 18cbd38c authored by Yann Garcia's avatar Yann Garcia
Browse files

Add oneM2M MQTT support

parent 648fbb1f

go-apps/meep-sss/Dockerfile

+4 −0
Original line number Diff line number Diff line
@@ -20,4 +20,8 @@ COPY ./data / 


RUN chmod +x /entrypoint.sh



RUN dpkg --configure -a



EXPOSE 33122/tcp



ENTRYPOINT ["/entrypoint.sh"]

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

+1 −9
Original line number Diff line number Diff line
@@ -180,15 +180,7 @@ func (broker_mqtt *SssMgrMqtt) send(p_ctx SssMgrBindingProtocolContext) (err err 
		d := make(map[string]int, 0)

		for k, v := range p_ctx.queries {

			if k == "ty" { // mosquitto_pub -d -q 0 -h 172.29.10.56 -p 1883 -t "/oneM2M/req/CAdmin/laboai-acme-ic-cse/json" -m "{\"fr\":\"CAdmin\",\"op\":2,\"rqi\":\"432bb877-7dc5-4e4d-b424-9c0d50604596\",\"rvi\":\"4\",\"to\":\"laboai-cse-in/YannouDomainAutomation/YannouGardenZone0\",\"ty\":3,\"fc\":{\"fu\":2,\"fo\":1}}"

				//body[k] = v

				i, err := strconv.Atoi(v) // body[k], err := strconv.Atoi(v)

				body[k] = i               // + 1

				/**

				 * FIXME FSCOM Based on ACME, there is a different behavior between MQTT & HTTP DISCOVERY to get the list of contaimer.

				 * For HTTP, ty = 3 (AE !?)

				 * FOR MQTT, ty = 4 (CNT)

				 * This is the reason of the 'body[k] = i + 1' for MQTT and not for HTTP

				 */

				body[k], err = strconv.Atoi(v)

				if err != nil {

					log.Error(err.Error())

					return err, nil

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

+495 −266

File changed.

Preview size limit exceeded, changes collapsed.

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

+213 −58
Original line number Diff line number Diff line
@@ -18,7 +18,6 @@ package sssmgr 


import (

	"fmt"

	"math/rand"

	"reflect"

	"testing"
@@ -34,107 +33,202 @@ func TestNewSssMgr(t *testing.T) { 


	// Invalid Connector

	fmt.Println("Invalid SSS Asset Manager")

	tm, err := NewSssMgr("", tmNamespace, nil, nil, nil)

	tm, err := NewSssMgr("", tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

	if err == nil || tm != nil {

		t.Fatalf("DB connection should have failed")

		t.Fatalf("Service name not set")

	}

	tm, err = NewSssMgr(tmName, tmNamespace, "", "172.29.10.56", 1883, nil, nil, nil)

	if err == nil || tm != nil {

		t.Fatalf("Binding protocol not set")

	}

	tm, err = NewSssMgr(tmName, tmNamespace, "MQTT", "", 1883, nil, nil, nil)

	if err == nil || tm != nil {

		t.Fatalf("Binding protocol not set")

	}



	// Valid Connector

	fmt.Println("Create valid SSS Asset Manager")

	tm, err = NewSssMgr(tmName, tmNamespace, nil, nil, nil)

	tm, err = NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

	if err != nil || tm == nil {

		t.Fatalf("Failed to create SSS Asset Manager")

	}



	// Cleanup

	err = tm.DeleteSssMgr()

	if err != nil {

		t.Fatalf("Failed to cleanup SSS Asset Manager")

	}

}

	tm = nil



func TestPopulateDevicesPerIotPlatforms(t *testing.T) {

	fmt.Println("--- ", t.Name())

	log.MeepTextLogInit(t.Name())



	// Valid Connector

	fmt.Println("Create valid SSS Asset Manager")

	tm, err := NewSssMgr(tmName, tmNamespace, nil, nil, nil)

	tm, err = NewSssMgr(tmName, tmNamespace, "HTTP", "172.29.10.56", 1883, nil, nil, nil)

	if err != nil || tm == nil {

		t.Fatalf("Failed to create SSS Asset Manager")

	}



	err = tm.populateDevicesPerIotPlatforms()

	if err != nil {

		t.Fatalf(err.Error())

	}



	// Cleanup

	err = tm.DeleteSssMgr()

	if err != nil {

		t.Fatalf("Failed to cleanup SSS Asset Manager")

	}

	tm = nil

}



func TestSensorDiscoveryInfoAll(t *testing.T) {

	fmt.Println("--- ", t.Name())

	log.MeepTextLogInit(t.Name())

// func TestPopulateDevicesPerIotPlatformsHttp(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())

// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "HTTP", "", 0, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



	// Valid Connector

	fmt.Println("Create valid SSS Asset Manager")

	tm, err := NewSssMgr(tmName, tmNamespace, nil, nil, nil)

	if err != nil || tm == nil {

		t.Fatalf("Failed to create SSS Asset Manager")

	}

// 	err = tm.populateDevicesPerIotPlatforms()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}



	err = tm.populateDevicesPerIotPlatforms()

	if err != nil {

		t.Fatalf(err.Error())

	}

// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// 	tm = nil

// }



	sensors, err := tm.SensorDiscoveryInfoAll()

	if err != nil {

		t.Fatalf(err.Error())

	}

	fmt.Println("sensors: ", sensors)

// func TestSensorDiscoveryInfoAllHttp(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())

// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "HTTP", "", 0, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



	// Cleanup

	err = tm.DeleteSssMgr()

	if err != nil {

		t.Fatalf("Failed to cleanup SSS Asset Manager")

	}

}

// 	sensors, err := tm.SensorDiscoveryInfoAll()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}

// 	fmt.Println("len=", len(sensors))

// 	fmt.Println("sensors", sensors)



// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// 	tm = nil

// }



// func TestGetSensorHttp(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())



// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "HTTP", "", 0, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



// 	sensors, err := tm.SensorDiscoveryInfoAll()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}



// 	for _, v := range sensors {

// 		fmt.Println("v", v)

// 		fmt.Println("TypeOf(v)", reflect.TypeOf(v))



// 		sensor, err := tm.GetSensor(v.SensorIdentifier)

// 		if !validate_sensor_discovery_info(v, sensor) {

// 			t.Fatalf(err.Error())

// 		}

// 	}



// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// 	tm = nil

// }



func TestGetSensor(t *testing.T) {

// func TestPopulateDevicesPerIotPlatformsMqtt(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())

// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}

// 	err = tm.populateDevicesPerIotPlatforms()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}

// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// 	tm = nil

// }



// func TestSensorDiscoveryInfoAllMqtt(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())



// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



// 	err = tm.populateDevicesPerIotPlatforms()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}



// 	sensors, err := tm.SensorDiscoveryInfoAll()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}

// 	fmt.Println("sensors: ", sensors)



// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// }



func TestGetSensorMqtt(t *testing.T) {

	fmt.Println("--- ", t.Name())

	log.MeepTextLogInit(t.Name())



	// Valid Connector

	fmt.Println("Create valid SSS Asset Manager")

	tm, err := NewSssMgr(tmName, tmNamespace, nil, nil, nil)

	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

	if err != nil || tm == nil {

		t.Fatalf("Failed to create SSS Asset Manager")

	}



	err = tm.populateDevicesPerIotPlatforms()

	if err != nil {

		t.Fatalf(err.Error())

	}



	sensors, err := tm.SensorDiscoveryInfoAll()

	if err != nil {

		t.Fatalf(err.Error())

	}

	fmt.Println("sensors: ", sensors)



	idx := rand.Int31n(int32(len(sensors)))

	sensor, err := tm.GetSensor(sensors[idx].SensorIdentifier)

	if err != nil {

	for _, v := range sensors {

		fmt.Println("v", v)

		fmt.Println("TypeOf(v)", reflect.TypeOf(v))



		sensor, err := tm.GetSensor(v.SensorIdentifier)

		if !validate_sensor_discovery_info(v, sensor) {

			t.Fatalf(err.Error())

		}

	fmt.Println("sensor: ", sensor)

	if !validate_sensor_discovery_info(sensors[idx], sensor) {

		t.Fatalf("Value mismatch")

	}



	// Cleanup
@@ -142,15 +236,76 @@ func TestGetSensor(t *testing.T) { 
	if err != nil {

		t.Fatalf("Failed to cleanup SSS Asset Manager")

	}

	tm = nil

}



// func TestVaidateOneM2MNotificationServer(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())



// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



// 	tm.init()

// 	fmt.Println("Waiting for 2 minutes to do curl request: curl -v http://mec-platform.etsi.org:33122/sbxykqjr17/mep1/sens/v1 ")



// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// }



// func TestGetSensor(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())



// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}



// 	err = tm.populateDevicesPerIotPlatforms()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}



// 	sensors, err := tm.SensorDiscoveryInfoAll()

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}

// 	fmt.Println("sensors: ", sensors)



// 	idx := rand.Int31n(int32(len(sensors)))

// 	sensor, err := tm.GetSensor(sensors[idx].SensorIdentifier)

// 	if err != nil {

// 		t.Fatalf(err.Error())

// 	}

// 	fmt.Println("sensor: ", sensor)

// 	if !validate_sensor_discovery_info(sensors[idx], sensor) {

// 		t.Fatalf("Value mismatch")

// 	}



// 	// Cleanup

// 	err = tm.DeleteSssMgr()

// 	if err != nil {

// 		t.Fatalf("Failed to cleanup SSS Asset Manager")

// 	}

// }



// func TestOneM2mCreateAEAndCNT(t *testing.T) {

// 	fmt.Println("--- ", t.Name())

// 	log.MeepTextLogInit(t.Name())



// 	// Valid Connector

// 	fmt.Println("Create valid SSS Asset Manager")

// 	tm, err := NewSssMgr(tmName, tmNamespace, nil, nil, nil)

// 	tm, err := NewSssMgr(tmName, tmNamespace, "MQTT", "172.29.10.56", 1883, nil, nil, nil)

// 	if err != nil || tm == nil {

// 		t.Fatalf("Failed to create SSS Asset Manager")

// 	}

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

+3 −3
Original line number Diff line number Diff line
@@ -161,9 +161,9 @@ func (broker_mqtt *message_broker_mqtt) Send(tm *TrafficMgr, msgContent string, 
			log.Error(err.Error())

			return err

		}

		log.Info("message_broker_simu: Send: Publish content : ", content)

		log.Info("message_broker_simu: Send: msgEncodeFormat: ", msgEncodeFormat)

		log.Info("message_broker_simu: Send: stdOrganization: ", stdOrganization)

		log.Info("message_broker_mqtt: Send: Publish content : ", content)

		log.Info("message_broker_mqtt: Send: msgEncodeFormat: ", msgEncodeFormat)

		log.Info("message_broker_mqtt: Send: stdOrganization: ", stdOrganization)

		token := broker_mqtt.client.Publish(tm.topic, 0, false, content)

		token.Wait()