Fixed integration and subscription routes related issues for R3

[ric-plt/rtmgr.git] / pkg / sbi / sbi.go

diff --git a/pkg/sbi/sbi.go b/pkg/sbi/sbi.go
index a2da825..2737cc1 100644 (file)
--- a/pkg/sbi/sbi.go
+++ b/pkg/sbi/sbi.go
@@ -1,4 +1,5 @@
 /*
+w
 ==================================================================================
   Copyright (c) 2019 AT&T Intellectual Property.
   Copyright (c) 2019 Nokia
@@ -14,6 +15,10 @@
    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,51 +31,107 @@ package sbi
 
 import (
        "errors"
-       "fmt"
-       "rtmgr"
+       "routing-manager/pkg/rtmgr"
+       "strconv"
 )
 
+const DefaultNngPipelineSocketPrefix = "tcp://"
+const DefaultNngPipelineSocketNumber = 4561
+const PlatformType = "platform"
+
 var (
-       SupportedSbis = []*SbiEngineConfig{
-               &SbiEngineConfig{
-                       SbiEngine{
-                               Name:     "nngpub",
-                               Version:  "v1",
-                               Protocol: "nngpubsub",
-                       },
-                       openSocket(openNngPub),
-                       closeSocket(closeNngPub),
-                       distributeAll(publishAll),
-                       true,
-               },
-               &SbiEngineConfig{
-                       SbiEngine{
-                               Name:     "nngpush",
-                               Version:  "v1",
-                               Protocol: "nngpipeline",
-                       },
-                       openSocket(nil),
-                       closeSocket(nil),
-                       distributeAll(nil),
-                       false,
+       SupportedSbis = []*EngineConfig{
+               {
+                       Name:        "nngpush",
+                       Version:     "v1",
+                       Protocol:    "nngpipeline",
+                       Instance:    NewNngPush(),
+                       IsAvailable: true,
                },
        }
 )
 
-func ListSbis() {
-       fmt.Printf("SBI:\n")
+func GetSbi(sbiName string) (Engine, error) {
        for _, sbi := range SupportedSbis {
-               if sbi.IsAvailable {
-                       rtmgr.Logger.Info(sbi.Engine.Name + "/" + sbi.Engine.Version)
+               if sbi.Name == sbiName && sbi.IsAvailable {
+                       return sbi.Instance, nil
                }
        }
+       return nil, errors.New("SBI:" + sbiName + " is not supported or still not available")
 }
 
-func GetSbi(sbiName string) (*SbiEngineConfig, error) {
-       for _, sbi := range SupportedSbis {
-               if sbi.Engine.Name == sbiName && sbi.IsAvailable {
-                       return sbi, nil
+type Sbi struct {
+}
+
+func (s *Sbi) pruneEndpointList(sbi Engine) {
+       rtmgr.Logger.Debug("pruneEndpointList invoked.")
+       for _, ep := range rtmgr.Eps {
+               if !ep.Keepalive {
+                       rtmgr.Logger.Debug("deleting %v", ep)
+                       sbi.DeleteEndpoint(ep)
+                       delete(rtmgr.Eps, ep.Uuid)
+               } else {
+                       rtmgr.Eps[ep.Uuid].Keepalive = false
+               }
+       }
+}
+
+func (s *Sbi) updateEndpoints(rcs *rtmgr.RicComponents, sbi Engine) {
+       for _, xapp := range (*rcs).XApps {
+               for _, instance := range xapp.Instances {
+                       uuid := instance.Ip + ":" + strconv.Itoa(int(instance.Port))
+                       if _, ok := rtmgr.Eps[uuid]; ok {
+                               rtmgr.Eps[uuid].Keepalive = true
+                       } else {
+                               ep := &rtmgr.Endpoint{
+                                       Uuid:       uuid,
+                                       Name:       instance.Name,
+                                       XAppType:   xapp.Name,
+                                       Ip:         instance.Ip,
+                                       Port:       instance.Port,
+                                       TxMessages: instance.TxMessages,
+                                       RxMessages: instance.RxMessages,
+                                       Socket:     nil,
+                                       IsReady:    false,
+                                       Keepalive:  true,
+                               }
+                               if err := sbi.AddEndpoint(ep); err != nil {
+                                       rtmgr.Logger.Error("can't create socket for endpoint: " + ep.Name + " due to:" + err.Error())
+                                       continue
+                               }
+                               rtmgr.Eps[uuid] = ep
+                       }
+               }
+       }
+       s.updatePlatformEndpoints(&((*rcs).Pcs), sbi)
+       s.pruneEndpointList(sbi)
+}
+
+func (s *Sbi) updatePlatformEndpoints(pcs *rtmgr.PlatformComponents, sbi Engine) {
+       rtmgr.Logger.Debug("updatePlatformEndpoints invoked. PCS: %v", *pcs)
+       for _, pc := range *pcs {
+               uuid := pc.Fqdn + ":" + strconv.Itoa(int(pc.Port))
+               if _, ok := rtmgr.Eps[uuid]; ok {
+                       rtmgr.Eps[uuid].Keepalive = true
+               } else {
+                       ep := &rtmgr.Endpoint{
+                               Uuid:       uuid,
+                               Name:       pc.Name,
+                               XAppType:   PlatformType,
+                               Ip:         pc.Fqdn,
+                               Port:       pc.Port,
+                               TxMessages: rtmgr.PLATFORMMESSAGETYPES[pc.Name]["tx"],
+                               RxMessages: rtmgr.PLATFORMMESSAGETYPES[pc.Name]["rx"],
+                               Socket:     nil,
+                               IsReady:    false,
+                               Keepalive:  true,
+                       }
+                       rtmgr.Logger.Debug("ep created: %v", ep)
+                       if err := sbi.AddEndpoint(ep); err != nil {
+                               rtmgr.Logger.Error("can't create socket for endpoint: " + ep.Name + " due to:" + err.Error())
+                               continue
+                       }
+                       rtmgr.Eps[uuid] = ep
                }
        }
-       return nil, errors.New("SBI:" + sbiName + "is not supported or still not a available")
 }