Commit ad5bfa05 authored by Yann Garcia's avatar Yann Garcia
Browse files

Add MEC 040 subscription mechanisms

parent 6ae2e1df

go-apps/meep-federation/server/convert.go

+9 −0
Changes for go-apps/meep-federation/server/convert.go: 9 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -40,3 +40,12 @@ func convertSystemInfotoJson(systemInfo *SystemInfo) string { 
	}

	return string(jsonInfo)

}



func convertSystemUpdateNotificationSubscriptionToJson(systemEvent *SystemUpdateNotificationSubscription) string {

	jsonInfo, err := json.Marshal(*systemEvent)

	if err != nil {

		log.Error(err.Error())

		return ""

	}

	return string(jsonInfo)

}

go-apps/meep-federation/server/federation.go

+346 −45
Changes for go-apps/meep-federation/server/federation.go: 346 added lines, 45 removed lines.
Original line number Diff line number Diff line
@@ -17,10 +17,12 @@ 
package server



import (

	"bytes"

	"context"

	"encoding/json"

	"errors"

	"fmt"

	"io/ioutil"

	"net/http"

	"net/url"

	"os"
@@ -34,6 +36,7 @@ import ( 
	dkm "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-key-mgr"

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

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

	met "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-metrics"

	redis "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-redis"

	scc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-sandbox-ctrl-client"

	smc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-service-mgmt-client"
@@ -92,7 +95,8 @@ var sbxCtrlClient *scc.APIClient 
var registrationTicker *time.Ticker

var subMgr *sm.SubscriptionMgr = nil



var subscriptionExpiryMap = map[int][]int{}      // List of subscriptions

var systemUpdateNotificationSubscriptionMap = map[int]*SystemUpdateNotificationSubscription{}

var subscriptionExpiryMap = map[int64][]int{}    // List of subscriptions

var systemInfopMap = map[string]sbi.SystemInfo{} // List of discovered MEC Plateform

var localSystemInfo sbi.SystemInfo               // Local registration from MEO
@@ -100,6 +104,8 @@ var mutex sync.Mutex 
var expiryTicker *time.Ticker

var nextSubscriptionIdAvailable int



var metricStore *met.MetricStore



func getAppInstanceId() (id string, err error) {

	var appInfo scc.ApplicationInfo

	appInfo.Id = instanceId
@@ -576,7 +582,8 @@ func cleanUp() { 
	// Flush all service data

	rc.DBFlush(baseKey)



	subscriptionExpiryMap = map[int][]int{}

	subscriptionExpiryMap = map[int64][]int{}

	systemUpdateNotificationSubscriptionMap = map[int]*SystemUpdateNotificationSubscription{}



	// Reset metrics store name

	updateStoreName("")
@@ -598,6 +605,15 @@ func updateStoreName(storeName string) { 
			return

		}

	}



	// Connect to Metric Store

	var err error

	metricStore, err = met.NewMetricStore(storeName, sandboxName, influxAddr, redisAddr)

	if err != nil {

		log.Error("Failed connection to metric-store: ", err)

		return

	}



}



/*
@@ -839,31 +855,224 @@ func systeminfoPOST(w http.ResponseWriter, r *http.Request) { 
}



func subscriptionDELETE(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> subscriptionDELETE: ", r)



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

	w.WriteHeader(http.StatusOK)



	// Retrieve variable parameters

	vars := mux.Vars(r)

	log.Info("vars: ", vars)

	subscriptionId := vars["subscriptionId"]

	if subscriptionId == "" {

		err := errors.New("Wrong parameters")

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)

		return

	}

	log.Info("subscriptionId: ", subscriptionId)



	err := delSubscription(subscriptionId)

	if err != nil {

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)

		return

	}



	// Prepare & send response

	w.WriteHeader(http.StatusNoContent)

}



func subscriptionGET(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> subscriptionGET: ", r)



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



	// Retrieve variable parameters

	vars := mux.Vars(r)

	log.Info("vars: ", vars)

	subscriptionId := vars["subscriptionId"]

	if subscriptionId == "" {

		err := errors.New("Wrong parameters")

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusBadRequest)

		return

	}

	log.Info("subscriptionId: ", subscriptionId)



	keyName := baseKey + "subscriptions:" + subscriptionId

	log.Info("delSubscription: keyName: ", keyName)

	jsonRespDB, err := rc.JSONGetEntry(keyName, ".")

	if err != nil { // subscriptionId does not exist

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusNotFound)

		return

	}



	if jsonRespDB == "" {

		w.WriteHeader(http.StatusInternalServerError)

		return

	}



	var subscription SystemUpdateNotificationSubscription

	err = json.Unmarshal([]byte(jsonRespDB), &subscription)

	if err != nil {

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusInternalServerError)

		return

	}



	fmt.Fprint(w, string(jsonRespDB))

	w.WriteHeader(http.StatusOK)

}



func subscriptionPOST(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> subscriptionPOST")



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

	w.WriteHeader(http.StatusOK)



	// Read JSON input stream provided in the Request, and stores it in the bodyBytes as bytes

	var subscription SystemUpdateNotificationSubscription

	bodyBytes, _ := ioutil.ReadAll(r.Body)

	log.Info("subscriptionsPost: bodyBytes: ", string(bodyBytes))

	// Unmarshal function to converts a JSON-formatted string into a SubscriptionCommon struct and store it in extractSubType

	err := json.Unmarshal(bodyBytes, &subscription)

	if err != nil {

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusInternalServerError)

		return

	}

	log.Info("subscriptionsPost: subscription: ", subscription)



	// Validating mandatory parameters provided in the request body

	if subscription.SubscriptionType == "" {

		log.Error("Mandatory SubscriptionType parameter should be present")

		errHandlerProblemDetails(w, "Mandatory attribute SubscriptionType is missing in the request body.", http.StatusBadRequest)

		return

	}



	if subscription.SubscriptionType != "SystemUpdateNotificationSubscription" {

		log.Error("Invalid SubscriptionType")

		errHandlerProblemDetails(w, "Invalid SubscriptionType", http.StatusBadRequest)

		return

	}



	if subscription.CallbackReference == "" {

		log.Error("CallbackReference shall be present.")

		errHandlerProblemDetails(w, "CallbackReference shall be present.", http.StatusBadRequest)

		return

	}



	// Validating mandatory parameters provided in the request body

	if subscription.Links != nil {

		log.Error("Links attribute should not be present in request body")

		errHandlerProblemDetails(w, "Links attribute should not be present in request body", http.StatusBadRequest)

		return

	}



	// subscriptionId will be generated sequentially

	newSubsId := nextSubscriptionIdAvailable

	nextSubscriptionIdAvailable++

	subsIdStr := strconv.Itoa(newSubsId)



	// create a unique link for every subscription and concatenate subscription to it

	link := new(Links)

	self := new(LinkType)

	self.Href = hostUrl.String() + basePath + "subscriptions/" + subsIdStr

	link.Self = self

	subscription.Links = link



	registerSubscription(subsIdStr, &subscription)

	w.Header().Set("Location", subscription.Links.Self.Href)



	// Prepare response

	jsonResponse := convertSystemUpdateNotificationSubscriptionToJson(&subscription)

	if jsonResponse == "" {

		log.Error("Marshalling failure")

		errHandlerProblemDetails(w, "Marshalling failure", http.StatusInternalServerError)

		return

	}



	// Store subscription key in redis

	keyName := baseKey + "subscriptions:" + subsIdStr

	log.Info("subscriptionPOST: keyName: ", keyName)

	log.Info("subscriptionPOST: subscription: ", subscription)

	err = rc.JSONSetEntry(keyName, ".", jsonResponse)

	if err != nil {

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusInternalServerError)

		return

	}

	log.Info("subscriptionPOST: jsonResponse: ", jsonResponse)



	// Send response

	w.WriteHeader(http.StatusCreated)

	fmt.Fprint(w, jsonResponse)

}



func subscriptionPUT(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> subscriptionPUT: ", r)



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

	w.WriteHeader(http.StatusOK)

}



func subscriptionsGET(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> subscriptionsGET: ", r)

	log.Debug(">>> subscriptionsGET: systemUpdateNotificationSubscriptionMap: ", systemUpdateNotificationSubscriptionMap)



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



	u, _ := url.Parse(r.URL.String())

	log.Info("url: ", u.RequestURI())

	q := u.Query()

	subscriptionType := q.Get("subscriptionType")

	systemId := q.Get("systemId")



	noFilter := true

	if subscriptionType != "" || systemId != "" {

		noFilter = false

	}



	subscriptionLinkList := new(SubscriptionLinkList)

	link := new(Links)

	self := new(LinkType)

	self.Href = hostUrl.String() + basePath + "subscriptions"

	link.Self = self

	subscriptionLinkList.Links = link

	for _, v := range systemUpdateNotificationSubscriptionMap {

		if noFilter {

			subscriptionLinkList.Subscription = append(subscriptionLinkList.Subscription, SubscriptionLinkListSubscription{Href: v.Links.Self.Href, SubscriptionType: v.SubscriptionType})

		} else {

			if subscriptionType == "system_update_notification_subscription" {

				subscriptionLinkList.Subscription = append(subscriptionLinkList.Subscription, SubscriptionLinkListSubscription{Href: v.Links.Self.Href, SubscriptionType: v.SubscriptionType})

			}

			if systemId != "" {

				for _, s := range v.SystemId {

					if s == systemId {

						subscriptionLinkList.Subscription = append(subscriptionLinkList.Subscription, SubscriptionLinkListSubscription{Href: v.Links.Self.Href, SubscriptionType: v.SubscriptionType})

						break

					}

				}

			}

		}

	}

	log.Debug("subscriptionsGET: subscriptionLinkList: ", subscriptionLinkList)



	// Prepare and send the response

	jsonResponse, err := json.Marshal(subscriptionLinkList)

	if err != nil {

		log.Error(err.Error())

		errHandlerProblemDetails(w, err.Error(), http.StatusInternalServerError)

		return

	}

	log.Debug("subscriptionsGET: jsonResponse: ", jsonResponse)

	w.WriteHeader(http.StatusOK)

	fmt.Fprint(w, string(jsonResponse))

}



func serviceGET(w http.ResponseWriter, r *http.Request) {

	log.Debug(">>> serviceGET: ", r)



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

	w.WriteHeader(http.StatusOK)

}
@@ -999,9 +1208,132 @@ func fedNotify(msg string, systemId string) { 
		systemInfopMap[systemInfo.SystemId] = systemInfo

		log.Debug("fedNotify: new systemInfopMap: ", systemInfopMap)



		// TODO Send notification if any subscription

		log.Debug("fedNotify: Send notification: ", systemInfo)

		// Send notification if any subscription

		for _, v := range systemUpdateNotificationSubscriptionMap {

			if v.SystemId == nil || len(v.SystemId) == 0 {

				// Send notification to v.CallbackReference

				log.Info("Sending notification ", v.CallbackReference)

				var notif = SystemUpdateNotification{

					NotificationType: v.SubscriptionType,

					UpdatedSystemInfo: []SystemInfo{

						SystemInfo{

							SystemId:       systemInfo.SystemId,

							SystemName:     systemInfo.SystemName,

							SystemProvider: systemInfo.SystemProvider,

						},

					},

					Links: &Links1{

						Subscription: v.Links.Self,

					},

				}

				log.Debug("fedNotify: Send notification: ", notif)

				go sendNotification(v.CallbackReference, notif)

			}

		}

	}

}



func sendNotification(notifyUrl string, notification SystemUpdateNotification) {

	startTime := time.Now()

	jsonNotif, err := json.Marshal(notification)

	if err != nil {

		log.Error(err.Error())

	}



	resp, err := http.Post(notifyUrl, "application/json", bytes.NewBuffer(jsonNotif))

	duration := float64(time.Since(startTime).Microseconds()) / 1000.0

	_ = httpLog.LogNotification(notifyUrl, "POST", "", "", string(jsonNotif), resp, startTime)

	if err != nil {

		log.Error(err)

		met.ObserveNotification(sandboxName, serviceName, notification.NotificationType, notifyUrl, nil, duration)

		return

	}

	met.ObserveNotification(sandboxName, serviceName, notification.NotificationType, notifyUrl, resp, duration)

	defer resp.Body.Close()

}



func registerSubscription(subsIdStr string, eventSub *SystemUpdateNotificationSubscription) {

	log.Debug(">>> registerSubscription")



	subsId, err := strconv.Atoi(subsIdStr)

	if err != nil {

		log.Error(err)

	}



	mutex.Lock()

	defer mutex.Unlock()



	systemUpdateNotificationSubscriptionMap[subsId] = eventSub



	// Calculate expiry time by adding seconds to the current time

	if eventSub != nil && eventSub.ExpiryDeadline != nil {

		expiryTime := time.Now().Unix() + int64(eventSub.ExpiryDeadline.Seconds)

		// 1.1 Remove the existing one

		intList := subscriptionExpiryMap[expiryTime]

		// TODO FSCOM Common code with delSubscription, create a function to remove an expiruDeadline from the subscriptionExpiryMap

		for i, subsIndex := range intList {

			log.Info("registerSubscription: i: ", i)

			log.Info("registerSubscription: subsIndex: ", subsIndex)

			if subsIndex == subsId {

				log.Info("registerSubscription: found index, delete entry")

				// Remove item and update subscriptionExpiryMap

				subscriptionExpiryMap[expiryTime] = append(intList[:i], intList[i+1:]...)

				break

			}

		} // End of 'for' statement

		// 1.2 And add the new one

		subscriptionExpiryMap[expiryTime] = append(subscriptionExpiryMap[expiryTime], subsId)

	}



	log.Debug("registerSubscription: systemUpdateNotificationSubscriptionMap: ", systemUpdateNotificationSubscriptionMap)

	log.Debug("registerSubscription: subscriptionExpiryMap: ", subscriptionExpiryMap)

}



func delSubscription(subsIdStr string) error {

	log.Debug(">>> delSubscription: ", subsIdStr)



	subsId, err := strconv.Atoi(subsIdStr)

	if err != nil {

		log.Error(err)

	}



	mutex.Lock()

	defer mutex.Unlock()



	keyName := baseKey + "subscriptions:" + subsIdStr

	log.Info("delSubscription: keyName: ", keyName)

	_, err = rc.JSONGetEntry(keyName, ".")

	if err != nil { // subscriptionId does not exist

		log.Error(err.Error())

		return err

	}

	err = rc.JSONDelEntry(keyName, ".")

	if err != nil {

		log.Error(err.Error())

		return err

	}



	log.Info("delSubscription: Before removal: subscriptionExpiryMap: ", subscriptionExpiryMap)

	for i, subsIndexList := range subscriptionExpiryMap {

		log.Info("delSubscription: subsIndexList: ", subsIndexList)

		for j, subsIndex := range subsIndexList {

			log.Info("delSubscription: j: ", j)

			log.Info("delSubscription: subsIndex: ", subsIndex)

			if subsIndex == subsId {

				// FIXME FSCOM How to manage it subscriptionExpiryMap

				log.Info("delSubscription: found index, delete entry")

				subscriptionExpiryMap[i] = append(subscriptionExpiryMap[i][:j], subscriptionExpiryMap[i][j+1:]...)

				break

			}

		} // End of 'for' statement

	} // End of 'for' statement

	log.Info("delSubscription: After removal: subscriptionExpiryMap: ", subscriptionExpiryMap)



	log.Info("delSubscription: Befor systemUpdateNotificationSubscriptionMap", systemUpdateNotificationSubscriptionMap)

	delete(systemUpdateNotificationSubscriptionMap, subsId)

	log.Info("delSubscription: After systemUpdateNotificationSubscriptionMap", systemUpdateNotificationSubscriptionMap)



	return nil

}



/*
@@ -1011,7 +1343,7 @@ func checkForExpiredSubscriptions() { 
	//log.Debug(">>> checkForExpiredSubscriptions")

	// log.Info("checkForExpiredSubscriptions: fedMsgSubscriptionMap: ", fedMsgSubscriptionMap)



	nowTime := int(time.Now().Unix())

	nowTime := time.Now().Unix()

	mutex.Lock()

	defer mutex.Unlock()

	for expiryTime, subsIndexList := range subscriptionExpiryMap {
@@ -1021,44 +1353,13 @@ func checkForExpiredSubscriptions() { 
				subsIdStr := strconv.Itoa(subsId)

				keyName := baseKey + "subscriptions:" + subsIdStr

				log.Info("checkForExpiredSubscriptions: keyName: ", keyName)

				// FIXME FSCOM To be done

				// subscription, err := rc.JSONGetEntry(keyName, ".")

				// if err != nil {

				// 	log.Error(err.Error())

				// 	continue

				// }

				// cbRef := ""

				// if strings.Contains(subscription, PROV_CHG_UU_UNI) {

				// 	if provChgUuUniSubscriptionMap[subsId] != nil {

				// 		cbRef = provChgUuUniSubscriptionMap[subsId].CallbackReference

				// 	} else {

				// 		continue

				// 	}

				// } else if strings.Contains(subscription, PROV_CHG_UU_MBMS) {

				// 	if provChgUuMbmsSubscriptionMap[subsId] != nil {

				// 		cbRef = provChgUuMbmsSubscriptionMap[subsId].CallbackReference

				// 	} else {

				// 		continue

				// 	}

				// } else if strings.Contains(subscription, PROV_CHG_PC5) {

				// 	if provChgPc5SubscriptionMap[subsId] != nil {

				// 		cbRef = provChgPc5SubscriptionMap[subsId].CallbackReference

				// 	} else {

				// 		continue

				// 	}

				// } else if strings.Contains(subscription, V2X_MSG) {

				// 	if fedMsgSubscriptionMap[subsId] != nil {

				// 		cbRef = fedMsgSubscriptionMap[subsId].CallbackReference

				// 	} else {

				// 		continue

				// 	}

				// 	// FIXME FSCOM There is no PredQosNotification defined by the standard ETSI GS MEC 030 V3.1.1 (2023-03)

				// 	// } else if strings.Contains(subscription, PRED_QOS) {

				// 	// 	if predQosSubscriptionMap[subsId] != nil {

				// 	// 		sendExpiryPredQosSubscription(predQosSubscriptionMap[subsId])

				// 	// 	} else {

				// 	// 		continue

				// 	// 	}

				// }

				//cbRef := systemUpdateNotificationSubscriptionMap[subsId].CallbackReference



				// var notif ExpiryNotification
@@ -1072,11 +1373,11 @@ func checkForExpiredSubscriptions() { 
				// notif.ExpiryDeadline = &expiryTimeStamp

				// sendExpiryNotification(link.Subscription.Href, notif)



				// // Delete subscription

				// err = delSubscription(subsIdStr, "", true)

				// if err != nil {

				// 	log.Error(err.Error())

				// }

				// Delete subscription

				err := delSubscription(subsIdStr)

				if err != nil {

					log.Error(err.Error())

				}

			} // End of 'for' statement

		}

	} // End of 'for' statement