Loading go-apps/meep-ams/api/swagger.yaml +19 −16 Original line number Diff line number Diff line Loading @@ -916,6 +916,8 @@ components: x-etsi-mec-cardinality: '1' x-etsi-mec-origin-type: String adjacentAppInfo: items: type: object description: Identifiers to associate the information of target application instance. properties: appInstanceId: Loading @@ -929,10 +931,11 @@ components: # x-etsi-mec-cardinality: 0..1 # x-etsi-mec-origin-type: CommunicationInterface $ref: '#/components/schemas/CommunicationInterface' type: object x-etsi-mec-cardinality: 0..1 x-etsi-mec-cardinality: 0..N required: - appInstanceId minItems: 0 type: array timeStamp: # description: Time stamp. # x-etsi-mec-cardinality: 0..1 Loading go-apps/meep-ams/go.mod +1 −1 Original line number Diff line number Diff line Loading @@ -5,7 +5,7 @@ go 1.12 require ( github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-app-support-client v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-key-mgr v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model v0.0.0 // indirect github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-gis-cache v0.0.0 // indirect github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-http-logger v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger v0.0.0 Loading go-apps/meep-ams/sbi/ams-sbi.go 0 → 100755 +296 −0 Original line number Diff line number Diff line /* * Copyright (c) 2019 InterDigital Communications, Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package sbi import ( "errors" "sync" dataModel "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mod "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-model" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" sam "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-swagger-api-mgr" ) type SbiCfg struct { ModuleName string SandboxName string MepName string RedisAddr string Locality []string DeviceInfoCb func(string, string, []string) ScenarioNameCb func(string) CleanUpCb func() } type AmsSbi struct { moduleName string sandboxName string mepName string localityEnabled bool locality map[string]bool mqLocal *mq.MsgQueue handlerId int apiMgr *sam.SwaggerApiMgr activeModel *mod.Model updateDeviceInfoCB func(string, string, []string) updateScenarioNameCB func(string) cleanUpCB func() mutex sync.Mutex } var sbi *AmsSbi // Init - Location Service SBI initialization func Init(cfg SbiCfg) (err error) { // Create new SBI instance sbi = new(AmsSbi) sbi.moduleName = cfg.ModuleName sbi.sandboxName = cfg.SandboxName sbi.mepName = cfg.MepName sbi.updateDeviceInfoCB = cfg.DeviceInfoCb sbi.updateScenarioNameCB = cfg.ScenarioNameCb sbi.cleanUpCB = cfg.CleanUpCb // Fill locality map if len(cfg.Locality) > 0 { sbi.locality = make(map[string]bool) for _, locality := range cfg.Locality { sbi.locality[locality] = true } sbi.localityEnabled = true } else { sbi.localityEnabled = false } // Create message queue sbi.mqLocal, err = mq.NewMsgQueue(mq.GetLocalName(sbi.sandboxName), sbi.moduleName, sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to create Message Queue with error: ", err) return err } log.Info("Message Queue created") // Create Swagger API Manager sbi.apiMgr, err = sam.NewSwaggerApiMgr(sbi.moduleName, sbi.sandboxName, sbi.mepName, sbi.mqLocal) if err != nil { log.Error("Failed to create Swagger API Manager. Error: ", err) return err } log.Info("Swagger API Manager created") // Create new active scenario model modelCfg := mod.ModelCfg{ Name: "activeScenario", Namespace: sbi.sandboxName, Module: sbi.moduleName, UpdateCb: nil, DbAddr: cfg.RedisAddr, } sbi.activeModel, err = mod.NewModel(modelCfg) if err != nil { log.Error("Failed to create model: ", err.Error()) return err } log.Info("Active Scenario Model created") // Initialize service processActiveScenarioUpdate() return nil } // Run - MEEP Location Service execution func Run() (err error) { // Start Swagger API Manager (provider) err = sbi.apiMgr.Start(true, false) if err != nil { log.Error("Failed to start Swagger API Manager with error: ", err.Error()) return err } log.Info("Swagger API Manager started") // Add module Swagger APIs err = sbi.apiMgr.AddApis() if err != nil { log.Error("Failed to add Swagger APIs with error: ", err.Error()) return err } log.Info("Swagger APIs successfully added") // Register Message Queue handler handler := mq.MsgHandler{Handler: msgHandler, UserData: nil} sbi.handlerId, err = sbi.mqLocal.RegisterHandler(handler) if err != nil { log.Error("Failed to register local Msg Queue listener: ", err.Error()) return err } log.Info("Registered local Msg Queue listener") return nil } func Stop() (err error) { if sbi == nil { return } if sbi.mqLocal != nil { sbi.mqLocal.UnregisterHandler(sbi.handlerId) } if sbi.apiMgr != nil { // Remove APIs err = sbi.apiMgr.RemoveApis() if err != nil { log.Error("Failed to remove APIs with err: ", err.Error()) return err } } return nil } // Message Queue handler func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgScenarioActivate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioUpdate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioTerminate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioTerminate() default: log.Trace("Ignoring unsupported message: ", mq.PrintMsg(msg)) } } func processActiveScenarioTerminate() { log.Debug("processActiveScenarioTerminate") // Sync with active scenario store sbi.activeModel.UpdateScenario() sbi.cleanUpCB() } func processActiveScenarioUpdate() { sbi.mutex.Lock() defer sbi.mutex.Unlock() log.Debug("processActiveScenarioUpdate") // Get previous list of connected UEs prevUeNames := []string{} prevUeNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range prevUeNameList { if isUeConnected(name) { prevUeNames = append(prevUeNames, name) } } // Sync with active scenario store sbi.activeModel.UpdateScenario() scenarioName := sbi.activeModel.GetScenarioName() sbi.updateScenarioNameCB(scenarioName) // Update UE info ueNames := []string{} ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality procList := []string{} zone, procMap, err := getZoneProcMap(name) if err != nil { log.Error(err.Error()) continue } for _, procName := range procMap { procList = append(procList, procName) } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } // Add UE to list of valid UEs ueNames = append(ueNames, name) sbi.updateDeviceInfoCB(name, zone, procList) } // Update UEs that were removed (no longer in locality) for _, prevUeName := range prevUeNames { found := false for _, ueName := range ueNames { if ueName == prevUeName { found = true break } } if !found { sbi.updateDeviceInfoCB(prevUeName, "", nil) } } // find all apps under our mep } func getZoneProcMap(name string) (zone string, procMap map[string]string, err error) { ctx := sbi.activeModel.GetNodeContext(name) if ctx == nil { err = errors.New("Error getting context for: " + name) return } zone = ctx.Parents[mod.Zone] procMap = ctx.Children[mod.Proc] return zone, procMap, nil } func isUeConnected(name string) bool { node := sbi.activeModel.GetNode(name) if node != nil { pl := node.(*dataModel.PhysicalLocation) if pl.Connected { return true } } return false } func isInLocality(zone string) bool { if sbi.localityEnabled { if _, found := sbi.locality[zone]; !found { return false } } return true } go-apps/meep-ams/sbi/ams-sbi.goback 0 → 100755 +359 −0 Original line number Diff line number Diff line /* * Copyright (c) 2019 InterDigital Communications, Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package sbi import ( "errors" "sync" "time" dataModel "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model" gc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-gis-cache" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mod "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-model" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" sam "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-swagger-api-mgr" ) type SbiCfg struct { ModuleName string SandboxName string MepName string RedisAddr string Locality []string DeviceInfoCb func(string, string) ScenarioNameCb func(string) CleanUpCb func() } type AmsSbi struct { moduleName string sandboxName string mepName string localityEnabled bool locality map[string]bool mqLocal *mq.MsgQueue handlerId int apiMgr *sam.SwaggerApiMgr activeModel *mod.Model gisCache *gc.GisCache refreshTicker *time.Ticker updateDeviceInfoCB func(string, string) updateScenarioNameCB func(string) cleanUpCB func() mutex sync.Mutex } var sbi *AmsSbi // Init - Location Service SBI initialization func Init(cfg SbiCfg) (err error) { // Create new SBI instance sbi = new(AmsSbi) sbi.moduleName = cfg.ModuleName sbi.sandboxName = cfg.SandboxName sbi.mepName = cfg.MepName sbi.updateDeviceInfoCB = cfg.DeviceInfoCb sbi.updateScenarioNameCB = cfg.ScenarioNameCb sbi.cleanUpCB = cfg.CleanUpCb // Fill locality map if len(cfg.Locality) > 0 { sbi.locality = make(map[string]bool) for _, locality := range cfg.Locality { sbi.locality[locality] = true } sbi.localityEnabled = true } else { sbi.localityEnabled = false } // Create message queue sbi.mqLocal, err = mq.NewMsgQueue(mq.GetLocalName(sbi.sandboxName), sbi.moduleName, sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to create Message Queue with error: ", err) return err } log.Info("Message Queue created") // Create Swagger API Manager sbi.apiMgr, err = sam.NewSwaggerApiMgr(sbi.moduleName, sbi.sandboxName, sbi.mepName, sbi.mqLocal) if err != nil { log.Error("Failed to create Swagger API Manager. Error: ", err) return err } log.Info("Swagger API Manager created") // Create new active scenario model modelCfg := mod.ModelCfg{ Name: "activeScenario", Namespace: sbi.sandboxName, Module: sbi.moduleName, UpdateCb: nil, DbAddr: cfg.RedisAddr, } sbi.activeModel, err = mod.NewModel(modelCfg) if err != nil { log.Error("Failed to create model: ", err.Error()) return err } log.Info("Active Scenario Model created") // Connect to GIS cache sbi.gisCache, err = gc.NewGisCache(sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to GIS Cache: ", err.Error()) return err } log.Info("Connected to GIS Cache") // Initialize service processActiveScenarioUpdate() return nil } // Run - MEEP Location Service execution func Run() (err error) { // Start Swagger API Manager (provider) err = sbi.apiMgr.Start(true, false) if err != nil { log.Error("Failed to start Swagger API Manager with error: ", err.Error()) return err } log.Info("Swagger API Manager started") // Add module Swagger APIs err = sbi.apiMgr.AddApis() if err != nil { log.Error("Failed to add Swagger APIs with error: ", err.Error()) return err } log.Info("Swagger APIs successfully added") // Register Message Queue handler handler := mq.MsgHandler{Handler: msgHandler, UserData: nil} sbi.handlerId, err = sbi.mqLocal.RegisterHandler(handler) if err != nil { log.Error("Failed to register local Msg Queue listener: ", err.Error()) return err } log.Info("Registered local Msg Queue listener") // Start refresh loop //startRefreshTicker() return nil } func Stop() (err error) { if sbi == nil { return } // Stop refresh loop stopRefreshTicker() if sbi.mqLocal != nil { sbi.mqLocal.UnregisterHandler(sbi.handlerId) } if sbi.apiMgr != nil { // Remove APIs err = sbi.apiMgr.RemoveApis() if err != nil { log.Error("Failed to remove APIs with err: ", err.Error()) return err } } return nil } func startRefreshTicker() { log.Debug("Starting refresh loop") sbi.refreshTicker = time.NewTicker(1000 * time.Millisecond) go func() { for range sbi.refreshTicker.C { refreshPositions() } }() } func stopRefreshTicker() { if sbi.refreshTicker != nil { sbi.refreshTicker.Stop() sbi.refreshTicker = nil log.Debug("Refresh loop stopped") } } // Message Queue handler func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgScenarioActivate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioUpdate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioTerminate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioTerminate() default: log.Trace("Ignoring unsupported message: ", mq.PrintMsg(msg)) } } func processActiveScenarioTerminate() { log.Debug("processActiveScenarioTerminate") // Sync with active scenario store sbi.activeModel.UpdateScenario() sbi.cleanUpCB() } func processActiveScenarioUpdate() { sbi.mutex.Lock() defer sbi.mutex.Unlock() log.Debug("processActiveScenarioUpdate") // Get previous list of connected UEs prevUeNames := []string{} prevUeNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range prevUeNameList { if isUeConnected(name) { prevUeNames = append(prevUeNames, name) } } // Sync with active scenario store sbi.activeModel.UpdateScenario() scenarioName := sbi.activeModel.GetScenarioName() log.Info("SIMON ", scenarioName) sbi.updateScenarioNameCB(scenarioName) // Update UE info ueNames := []string{} ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { log.Info("SIMON check ", name) // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality zone, _, err := getNetworkLocation(name) if err != nil { log.Error(err.Error()) continue } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } // Add UE to list of valid UEs ueNames = append(ueNames, name) log.Info("SIMON ue ", name, "---", zone) sbi.updateDeviceInfoCB(name, zone) } // Update UEs that were removed (no longer in locality) for _, prevUeName := range prevUeNames { found := false for _, ueName := range ueNames { if ueName == prevUeName { found = true break } } if !found { sbi.updateDeviceInfoCB(prevUeName, "") log.Info("SIMON ue rm ", prevUeName) // log.Info("Ue removed : ", prevUeName) } } } func getNetworkLocation(name string) (zone string, netLoc string, err error) { ctx := sbi.activeModel.GetNodeContext(name) if ctx == nil { err = errors.New("Error getting context for: " + name) return } zone = ctx.Parents[mod.Zone] netLoc = ctx.Parents[mod.NetLoc] return zone, netLoc, nil } func refreshPositions() { sbi.mutex.Lock() defer sbi.mutex.Unlock() // Update UE Positions ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality zone, _, err := getNetworkLocation(name) if err != nil { log.Error(err.Error()) return } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } sbi.updateDeviceInfoCB(name, zone) } } func isUeConnected(name string) bool { node := sbi.activeModel.GetNode(name) if node != nil { pl := node.(*dataModel.PhysicalLocation) if pl.Connected { return true } } return false } func isInLocality(zone string) bool { if sbi.localityEnabled { if _, found := sbi.locality[zone]; !found { return false } } return true } go-apps/meep-ams/server/README.md +1 −1 Original line number Diff line number Diff line Loading @@ -13,7 +13,7 @@ To see how to make this your own, look here: [README](https://github.com/swagger-api/swagger-codegen/blob/master/README.md) - API version: 2.1.1 - Build date: 2021-09-14T17:37:11.354659-04:00[America/Toronto] - Build date: 2021-09-15T14:04:58.961-04:00[America/New_York] ### Running the server Loading Loading
go-apps/meep-ams/api/swagger.yaml +19 −16 Original line number Diff line number Diff line Loading @@ -916,6 +916,8 @@ components: x-etsi-mec-cardinality: '1' x-etsi-mec-origin-type: String adjacentAppInfo: items: type: object description: Identifiers to associate the information of target application instance. properties: appInstanceId: Loading @@ -929,10 +931,11 @@ components: # x-etsi-mec-cardinality: 0..1 # x-etsi-mec-origin-type: CommunicationInterface $ref: '#/components/schemas/CommunicationInterface' type: object x-etsi-mec-cardinality: 0..1 x-etsi-mec-cardinality: 0..N required: - appInstanceId minItems: 0 type: array timeStamp: # description: Time stamp. # x-etsi-mec-cardinality: 0..1 Loading
go-apps/meep-ams/go.mod +1 −1 Original line number Diff line number Diff line Loading @@ -5,7 +5,7 @@ go 1.12 require ( github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-app-support-client v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-key-mgr v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model v0.0.0 // indirect github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-gis-cache v0.0.0 // indirect github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-http-logger v0.0.0 github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger v0.0.0 Loading
go-apps/meep-ams/sbi/ams-sbi.go 0 → 100755 +296 −0 Original line number Diff line number Diff line /* * Copyright (c) 2019 InterDigital Communications, Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package sbi import ( "errors" "sync" dataModel "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mod "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-model" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" sam "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-swagger-api-mgr" ) type SbiCfg struct { ModuleName string SandboxName string MepName string RedisAddr string Locality []string DeviceInfoCb func(string, string, []string) ScenarioNameCb func(string) CleanUpCb func() } type AmsSbi struct { moduleName string sandboxName string mepName string localityEnabled bool locality map[string]bool mqLocal *mq.MsgQueue handlerId int apiMgr *sam.SwaggerApiMgr activeModel *mod.Model updateDeviceInfoCB func(string, string, []string) updateScenarioNameCB func(string) cleanUpCB func() mutex sync.Mutex } var sbi *AmsSbi // Init - Location Service SBI initialization func Init(cfg SbiCfg) (err error) { // Create new SBI instance sbi = new(AmsSbi) sbi.moduleName = cfg.ModuleName sbi.sandboxName = cfg.SandboxName sbi.mepName = cfg.MepName sbi.updateDeviceInfoCB = cfg.DeviceInfoCb sbi.updateScenarioNameCB = cfg.ScenarioNameCb sbi.cleanUpCB = cfg.CleanUpCb // Fill locality map if len(cfg.Locality) > 0 { sbi.locality = make(map[string]bool) for _, locality := range cfg.Locality { sbi.locality[locality] = true } sbi.localityEnabled = true } else { sbi.localityEnabled = false } // Create message queue sbi.mqLocal, err = mq.NewMsgQueue(mq.GetLocalName(sbi.sandboxName), sbi.moduleName, sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to create Message Queue with error: ", err) return err } log.Info("Message Queue created") // Create Swagger API Manager sbi.apiMgr, err = sam.NewSwaggerApiMgr(sbi.moduleName, sbi.sandboxName, sbi.mepName, sbi.mqLocal) if err != nil { log.Error("Failed to create Swagger API Manager. Error: ", err) return err } log.Info("Swagger API Manager created") // Create new active scenario model modelCfg := mod.ModelCfg{ Name: "activeScenario", Namespace: sbi.sandboxName, Module: sbi.moduleName, UpdateCb: nil, DbAddr: cfg.RedisAddr, } sbi.activeModel, err = mod.NewModel(modelCfg) if err != nil { log.Error("Failed to create model: ", err.Error()) return err } log.Info("Active Scenario Model created") // Initialize service processActiveScenarioUpdate() return nil } // Run - MEEP Location Service execution func Run() (err error) { // Start Swagger API Manager (provider) err = sbi.apiMgr.Start(true, false) if err != nil { log.Error("Failed to start Swagger API Manager with error: ", err.Error()) return err } log.Info("Swagger API Manager started") // Add module Swagger APIs err = sbi.apiMgr.AddApis() if err != nil { log.Error("Failed to add Swagger APIs with error: ", err.Error()) return err } log.Info("Swagger APIs successfully added") // Register Message Queue handler handler := mq.MsgHandler{Handler: msgHandler, UserData: nil} sbi.handlerId, err = sbi.mqLocal.RegisterHandler(handler) if err != nil { log.Error("Failed to register local Msg Queue listener: ", err.Error()) return err } log.Info("Registered local Msg Queue listener") return nil } func Stop() (err error) { if sbi == nil { return } if sbi.mqLocal != nil { sbi.mqLocal.UnregisterHandler(sbi.handlerId) } if sbi.apiMgr != nil { // Remove APIs err = sbi.apiMgr.RemoveApis() if err != nil { log.Error("Failed to remove APIs with err: ", err.Error()) return err } } return nil } // Message Queue handler func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgScenarioActivate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioUpdate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioTerminate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioTerminate() default: log.Trace("Ignoring unsupported message: ", mq.PrintMsg(msg)) } } func processActiveScenarioTerminate() { log.Debug("processActiveScenarioTerminate") // Sync with active scenario store sbi.activeModel.UpdateScenario() sbi.cleanUpCB() } func processActiveScenarioUpdate() { sbi.mutex.Lock() defer sbi.mutex.Unlock() log.Debug("processActiveScenarioUpdate") // Get previous list of connected UEs prevUeNames := []string{} prevUeNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range prevUeNameList { if isUeConnected(name) { prevUeNames = append(prevUeNames, name) } } // Sync with active scenario store sbi.activeModel.UpdateScenario() scenarioName := sbi.activeModel.GetScenarioName() sbi.updateScenarioNameCB(scenarioName) // Update UE info ueNames := []string{} ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality procList := []string{} zone, procMap, err := getZoneProcMap(name) if err != nil { log.Error(err.Error()) continue } for _, procName := range procMap { procList = append(procList, procName) } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } // Add UE to list of valid UEs ueNames = append(ueNames, name) sbi.updateDeviceInfoCB(name, zone, procList) } // Update UEs that were removed (no longer in locality) for _, prevUeName := range prevUeNames { found := false for _, ueName := range ueNames { if ueName == prevUeName { found = true break } } if !found { sbi.updateDeviceInfoCB(prevUeName, "", nil) } } // find all apps under our mep } func getZoneProcMap(name string) (zone string, procMap map[string]string, err error) { ctx := sbi.activeModel.GetNodeContext(name) if ctx == nil { err = errors.New("Error getting context for: " + name) return } zone = ctx.Parents[mod.Zone] procMap = ctx.Children[mod.Proc] return zone, procMap, nil } func isUeConnected(name string) bool { node := sbi.activeModel.GetNode(name) if node != nil { pl := node.(*dataModel.PhysicalLocation) if pl.Connected { return true } } return false } func isInLocality(zone string) bool { if sbi.localityEnabled { if _, found := sbi.locality[zone]; !found { return false } } return true }
go-apps/meep-ams/sbi/ams-sbi.goback 0 → 100755 +359 −0 Original line number Diff line number Diff line /* * Copyright (c) 2019 InterDigital Communications, Inc * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * * http://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. */ package sbi import ( "errors" "sync" "time" dataModel "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-data-model" gc "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-gis-cache" log "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-logger" mod "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-model" mq "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-mq" sam "github.com/InterDigitalInc/AdvantEDGE/go-packages/meep-swagger-api-mgr" ) type SbiCfg struct { ModuleName string SandboxName string MepName string RedisAddr string Locality []string DeviceInfoCb func(string, string) ScenarioNameCb func(string) CleanUpCb func() } type AmsSbi struct { moduleName string sandboxName string mepName string localityEnabled bool locality map[string]bool mqLocal *mq.MsgQueue handlerId int apiMgr *sam.SwaggerApiMgr activeModel *mod.Model gisCache *gc.GisCache refreshTicker *time.Ticker updateDeviceInfoCB func(string, string) updateScenarioNameCB func(string) cleanUpCB func() mutex sync.Mutex } var sbi *AmsSbi // Init - Location Service SBI initialization func Init(cfg SbiCfg) (err error) { // Create new SBI instance sbi = new(AmsSbi) sbi.moduleName = cfg.ModuleName sbi.sandboxName = cfg.SandboxName sbi.mepName = cfg.MepName sbi.updateDeviceInfoCB = cfg.DeviceInfoCb sbi.updateScenarioNameCB = cfg.ScenarioNameCb sbi.cleanUpCB = cfg.CleanUpCb // Fill locality map if len(cfg.Locality) > 0 { sbi.locality = make(map[string]bool) for _, locality := range cfg.Locality { sbi.locality[locality] = true } sbi.localityEnabled = true } else { sbi.localityEnabled = false } // Create message queue sbi.mqLocal, err = mq.NewMsgQueue(mq.GetLocalName(sbi.sandboxName), sbi.moduleName, sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to create Message Queue with error: ", err) return err } log.Info("Message Queue created") // Create Swagger API Manager sbi.apiMgr, err = sam.NewSwaggerApiMgr(sbi.moduleName, sbi.sandboxName, sbi.mepName, sbi.mqLocal) if err != nil { log.Error("Failed to create Swagger API Manager. Error: ", err) return err } log.Info("Swagger API Manager created") // Create new active scenario model modelCfg := mod.ModelCfg{ Name: "activeScenario", Namespace: sbi.sandboxName, Module: sbi.moduleName, UpdateCb: nil, DbAddr: cfg.RedisAddr, } sbi.activeModel, err = mod.NewModel(modelCfg) if err != nil { log.Error("Failed to create model: ", err.Error()) return err } log.Info("Active Scenario Model created") // Connect to GIS cache sbi.gisCache, err = gc.NewGisCache(sbi.sandboxName, cfg.RedisAddr) if err != nil { log.Error("Failed to GIS Cache: ", err.Error()) return err } log.Info("Connected to GIS Cache") // Initialize service processActiveScenarioUpdate() return nil } // Run - MEEP Location Service execution func Run() (err error) { // Start Swagger API Manager (provider) err = sbi.apiMgr.Start(true, false) if err != nil { log.Error("Failed to start Swagger API Manager with error: ", err.Error()) return err } log.Info("Swagger API Manager started") // Add module Swagger APIs err = sbi.apiMgr.AddApis() if err != nil { log.Error("Failed to add Swagger APIs with error: ", err.Error()) return err } log.Info("Swagger APIs successfully added") // Register Message Queue handler handler := mq.MsgHandler{Handler: msgHandler, UserData: nil} sbi.handlerId, err = sbi.mqLocal.RegisterHandler(handler) if err != nil { log.Error("Failed to register local Msg Queue listener: ", err.Error()) return err } log.Info("Registered local Msg Queue listener") // Start refresh loop //startRefreshTicker() return nil } func Stop() (err error) { if sbi == nil { return } // Stop refresh loop stopRefreshTicker() if sbi.mqLocal != nil { sbi.mqLocal.UnregisterHandler(sbi.handlerId) } if sbi.apiMgr != nil { // Remove APIs err = sbi.apiMgr.RemoveApis() if err != nil { log.Error("Failed to remove APIs with err: ", err.Error()) return err } } return nil } func startRefreshTicker() { log.Debug("Starting refresh loop") sbi.refreshTicker = time.NewTicker(1000 * time.Millisecond) go func() { for range sbi.refreshTicker.C { refreshPositions() } }() } func stopRefreshTicker() { if sbi.refreshTicker != nil { sbi.refreshTicker.Stop() sbi.refreshTicker = nil log.Debug("Refresh loop stopped") } } // Message Queue handler func msgHandler(msg *mq.Msg, userData interface{}) { switch msg.Message { case mq.MsgScenarioActivate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioUpdate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioUpdate() case mq.MsgScenarioTerminate: log.Debug("RX MSG: ", mq.PrintMsg(msg)) processActiveScenarioTerminate() default: log.Trace("Ignoring unsupported message: ", mq.PrintMsg(msg)) } } func processActiveScenarioTerminate() { log.Debug("processActiveScenarioTerminate") // Sync with active scenario store sbi.activeModel.UpdateScenario() sbi.cleanUpCB() } func processActiveScenarioUpdate() { sbi.mutex.Lock() defer sbi.mutex.Unlock() log.Debug("processActiveScenarioUpdate") // Get previous list of connected UEs prevUeNames := []string{} prevUeNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range prevUeNameList { if isUeConnected(name) { prevUeNames = append(prevUeNames, name) } } // Sync with active scenario store sbi.activeModel.UpdateScenario() scenarioName := sbi.activeModel.GetScenarioName() log.Info("SIMON ", scenarioName) sbi.updateScenarioNameCB(scenarioName) // Update UE info ueNames := []string{} ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { log.Info("SIMON check ", name) // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality zone, _, err := getNetworkLocation(name) if err != nil { log.Error(err.Error()) continue } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } // Add UE to list of valid UEs ueNames = append(ueNames, name) log.Info("SIMON ue ", name, "---", zone) sbi.updateDeviceInfoCB(name, zone) } // Update UEs that were removed (no longer in locality) for _, prevUeName := range prevUeNames { found := false for _, ueName := range ueNames { if ueName == prevUeName { found = true break } } if !found { sbi.updateDeviceInfoCB(prevUeName, "") log.Info("SIMON ue rm ", prevUeName) // log.Info("Ue removed : ", prevUeName) } } } func getNetworkLocation(name string) (zone string, netLoc string, err error) { ctx := sbi.activeModel.GetNodeContext(name) if ctx == nil { err = errors.New("Error getting context for: " + name) return } zone = ctx.Parents[mod.Zone] netLoc = ctx.Parents[mod.NetLoc] return zone, netLoc, nil } func refreshPositions() { sbi.mutex.Lock() defer sbi.mutex.Unlock() // Update UE Positions ueNameList := sbi.activeModel.GetNodeNames("UE") for _, name := range ueNameList { // Ignore disconnected UEs if !isUeConnected(name) { continue } // Get UE locality zone, _, err := getNetworkLocation(name) if err != nil { log.Error(err.Error()) return } // Ignore UEs in zones outside locality if !isInLocality(zone) { continue } sbi.updateDeviceInfoCB(name, zone) } } func isUeConnected(name string) bool { node := sbi.activeModel.GetNode(name) if node != nil { pl := node.(*dataModel.PhysicalLocation) if pl.Connected { return true } } return false } func isInLocality(zone string) bool { if sbi.localityEnabled { if _, found := sbi.locality[zone]; !found { return false } } return true }
go-apps/meep-ams/server/README.md +1 −1 Original line number Diff line number Diff line Loading @@ -13,7 +13,7 @@ To see how to make this your own, look here: [README](https://github.com/swagger-api/swagger-codegen/blob/master/README.md) - API version: 2.1.1 - Build date: 2021-09-14T17:37:11.354659-04:00[America/Toronto] - Build date: 2021-09-15T14:04:58.961-04:00[America/New_York] ### Running the server Loading
