X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=cmd%2Frtmgr.go;h=63b67b739cb85b647e0a486adbd0f41b7b1f5a2f;hb=488200cbf62e6dc51d1e5cc692c1e52e9ef28cb5;hp=e4f76d8630d31c08eefd236d902399d955a3845d;hpb=871fa393844ce1b61b8d5218d27687d9fc05803a;p=ric-plt%2Frtmgr.git

diff --git a/cmd/rtmgr.go b/cmd/rtmgr.go
index e4f76d8..63b67b7 100644
--- a/cmd/rtmgr.go
+++ b/cmd/rtmgr.go
@@ -14,6 +14,12 @@
    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.
+
+
+   This source code is part of the near-RT RIC (RAN Intelligent Controller)
+   platform project (RICP).
+
+
 ==================================================================================
 */
 /*
@@ -26,100 +32,156 @@
 */
 package main
 
+//TODO: change flag to pflag (won't need any argument parse)
+
 import (
-	"flag"
-	"nbi"
+	"gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/xapp"
 	"os"
-	"rpe"
-	"rtmgr"
-	"sbi"
-	"sdl"
+	"os/signal"
+	"routing-manager/pkg/nbi"
+	"routing-manager/pkg/rpe"
+	"routing-manager/pkg/rtmgr"
+	"routing-manager/pkg/sbi"
+	"routing-manager/pkg/sdl"
+	"syscall"
 	"time"
+	"sync"
 )
 
 const SERVICENAME = "rtmgr"
-const INTERVAL time.Duration = 2
-
-var (
-	args *map[string]string
-)
+const INTERVAL time.Duration = 60
 
-func parseArgs() {
-	a := make(map[string]string)
-	xmgeturl := flag.String("nbi-httpget", "http://localhost:3000/xapps", "xApp Manager URL")
-	nngpubsock := flag.String("sbi-nngsub", "tcp://0.0.0.0:4560", "NNG Subsciption Socket URI")
-	file := flag.String("sdl-file", "/db/rt.json", "Local file store location")
-	rpename := flag.String("rpe", "rmr", "Policy Engine Module name")
-	loglevel := flag.String("loglevel", "INFO", "INFO | WARN | ERROR | DEBUG")
-	flag.Parse()
-	if (*xmgeturl) != "" {
-		a["xmurl"] = (*xmgeturl)
-		a["nbiname"] = "httpGetter"
-	}
-	if (*nngpubsock) != "" {
-		a["socketuri"] = (*nngpubsock)
-		a["sbiname"] = "nngpub"
-	}
-	if (*file) != "" {
-		a["file"] = (*file)
-		a["sdlname"] = "file"
+func initRtmgr() (nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, err error) {
+	if nbiEngine, err = nbi.GetNbi(xapp.Config.GetString("nbi")); err == nil && nbiEngine != nil {
+		if sbiEngine, err = sbi.GetSbi(xapp.Config.GetString("sbi")); err == nil && sbiEngine != nil {
+			if sdlEngine, err = sdl.GetSdl(xapp.Config.GetString("sdl")); err == nil && sdlEngine != nil {
+				if rpeEngine, err = rpe.GetRpe(xapp.Config.GetString("rpe")); err == nil && rpeEngine != nil {
+					return nbiEngine, sbiEngine, sdlEngine, rpeEngine, nil
+				}
+			}
+		}
 	}
-	a["rpename"] = (*rpename)
-	a["loglevel"] = (*loglevel)
-	args = &a
+	return nil, nil, nil, nil, err
 }
 
-func initRtmgr() (*nbi.NbiEngineConfig, *sbi.SbiEngineConfig, *sdl.SdlEngineConfig, *rpe.RpeEngineConfig, error) {
-	var err error
-	if nbi, err := nbi.GetNbi((*args)["nbiname"]); err == nil && nbi != nil {
-		if sbi, err := sbi.GetSbi((*args)["sbiname"]); err == nil && sbi != nil {
-			if sdl, err := sdl.GetSdl((*args)["sdlname"]); err == nil && sdl != nil {
-				if rpe, err := rpe.GetRpe((*args)["rpename"]); err == nil && rpe != nil {
-					return nbi, sbi, sdl, rpe, nil
-				}
+
+
+func serveSBI(triggerSBI <-chan bool, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) {
+	for {
+		if <-triggerSBI {
+			m.Lock()
+			data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
+			m.Unlock()
+			if err != nil || data == nil {
+				xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
+				continue
+			}
+			sbiEngine.UpdateEndpoints(data)
+			policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
+			err = sbiEngine.DistributeAll(policies)
+			if err != nil {
+				xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
 			}
 		}
 	}
-	return nil, nil, nil, nil, err
 }
 
-func serve(nbi *nbi.NbiEngineConfig, sbi *sbi.SbiEngineConfig, sdl *sdl.SdlEngineConfig, rpe *rpe.RpeEngineConfig) {
-	err := sbi.OpenSocket((*args)["socketuri"])
+func sendRoutesToAll(sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine) {
+
+	data, err := sdlEngine.ReadAll(xapp.Config.GetString("rtfile"))
+	if err != nil || data == nil {
+		xapp.Logger.Error("Cannot get data from sdl interface due to: " + err.Error())
+		return
+	}
+	sbiEngine.UpdateEndpoints(data)
+	policies := rpeEngine.GeneratePolicies(rtmgr.Eps, data)
+	err = sbiEngine.DistributeAll(policies)
 	if err != nil {
-		rtmgr.Logger.Info("fail to open pub socket due to: " + err.Error())
+		xapp.Logger.Error("Routing table cannot be published due to: " + err.Error())
 		return
 	}
-	defer sbi.CloseSocket()
+}
+
+
+func serve(nbiEngine nbi.Engine, sbiEngine sbi.Engine, sdlEngine sdl.Engine, rpeEngine rpe.Engine, m *sync.Mutex) {
+
+	triggerSBI := make(chan bool)
+
+	nbiErr := nbiEngine.Initialize(xapp.Config.GetString("xmurl"), xapp.Config.GetString("nbiurl"), xapp.Config.GetString("rtfile"), xapp.Config.GetString("cfgfile"), xapp.Config.GetString("e2murl"), 
+		sdlEngine, rpeEngine, triggerSBI, m)
+	if nbiErr != nil {
+		xapp.Logger.Error("Failed to initialize nbi due to: " + nbiErr.Error())
+		return
+	}
+
+	err := sbiEngine.Initialize(xapp.Config.GetString("sbiurl"))
+	if err != nil {
+		xapp.Logger.Info("Failed to open push socket due to: " + err.Error())
+		return
+	}
+	defer nbiEngine.Terminate()
+	defer sbiEngine.Terminate()
+
+	// This SBI Go routine is trtiggered by periodic main loop and when data is recieved on REST interface.
+	go serveSBI(triggerSBI, sbiEngine, sdlEngine, rpeEngine, m)
+
 	for {
-		time.Sleep(INTERVAL * time.Second)
-		data, err := nbi.BatchFetch((*args)["xmurl"])
-		if err != nil {
-			rtmgr.Logger.Error("cannot get data from " + nbi.Engine.Name + " interface dute to: " + err.Error())
-		} else {
-			sdl.WriteAll((*args)["file"], data)
-		}
-		data, err = sdl.ReadAll((*args)["file"])
-		if err != nil || data == nil {
-			rtmgr.Logger.Error("cannot get data from " + sdl.Engine.Name + " interface dute to: " + err.Error())
-			continue
-		}
-		policies := rpe.GeneratePolicies(data)
-		err = sbi.DistributeAll(policies)
-		if err != nil {
-			rtmgr.Logger.Error("routing rable cannot be published due to: " + err.Error())
+		if xapp.Config.GetString("nbi") == "httpGetter" {
+			data, err := nbiEngine.(*nbi.HttpGetter).FetchAllXApps(xapp.Config.GetString("xmurl"))
+			if err != nil {
+				xapp.Logger.Error("Cannot fetch xapp data due to: " + err.Error())
+			} else if data != nil {
+				sdlEngine.WriteXApps(xapp.Config.GetString("rtfile"), data)
+			}
 		}
+
+		sendRoutesToAll(sbiEngine, sdlEngine, rpeEngine)
+
+		rtmgr.Rtmgr_ready = true
+		time.Sleep(INTERVAL * time.Second)
+		xapp.Logger.Debug("Periodic loop timed out. Setting triggerSBI flag to distribute updated routes.")
 	}
 }
 
+func SetupCloseHandler() {
+	c := make(chan os.Signal, 2)
+	signal.Notify(c, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-c
+		xapp.Logger.Info("\r- Ctrl+C pressed in Terminal")
+		os.Exit(0)
+	}()
+}
+
 func main() {
-	parseArgs()
-	rtmgr.SetLogLevel((*args)["loglevel"])
-	nbi, sbi, sdl, rpe, err := initRtmgr()
+
+	nbiEngine, sbiEngine, sdlEngine, rpeEngine, err := initRtmgr()
 	if err != nil {
-		rtmgr.Logger.Error(err.Error())
+		xapp.Logger.Error(err.Error())
 		os.Exit(1)
 	}
-	rtmgr.Logger.Info("Start " + SERVICENAME + " service")
-	serve(nbi, sbi, sdl, rpe)
+
+	SetupCloseHandler()
+
+	xapp.Logger.Info("Start " + SERVICENAME + " service")
+	rtmgr.Eps = make(rtmgr.Endpoints)
+	rtmgr.Rtmgr_ready = false
+
+	var m sync.Mutex
+
+// RMR thread is starting port: 4560
+	c := nbi.NewControl()
+	go c.Run(sbiEngine, sdlEngine, rpeEngine, &m)
+
+// Waiting for RMR to be ready
+	time.Sleep(time.Duration(2) * time.Second)
+	for xapp.Rmr.IsReady() == false {
+	        time.Sleep(time.Duration(2) * time.Second)
+	}
+
+	dummy_whid := int(xapp.Rmr.Openwh("localhost:4560"))
+	xapp.Logger.Info("created dummy Wormhole ID for routingmanager and dummy_whid :%d", dummy_whid)
+
+	serve(nbiEngine, sbiEngine, sdlEngine, rpeEngine, &m)
 	os.Exit(0)
 }