pkg/sbi/nngpush.go

   1 /*
   2 ==================================================================================
   3   Copyright (c) 2019 AT&T Intellectual Property.
   4   Copyright (c) 2019 Nokia
   5
   6    Licensed under the Apache License, Version 2.0 (the "License");
   7    you may not use this file except in compliance with the License.
   8    You may obtain a copy of the License at
   9
  10        http://www.apache.org/licenses/LICENSE-2.0
  11
  12    Unless required by applicable law or agreed to in writing, software
  13    distributed under the License is distributed on an "AS IS" BASIS,
  14    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15    See the License for the specific language governing permissions and
  16    limitations under the License.
  17 ==================================================================================
  18 */
  19 /*
  20   Mnemonic:     nngpipe.go
  21   Abstract: mangos (NNG) Pipeline SBI implementation
  22   Date:         12 March 2019
  23 */
  24
  25 package sbi
  26
  27 import (
  28         "errors"
  29         "nanomsg.org/go/mangos/v2"
  30         "nanomsg.org/go/mangos/v2/protocol/push"
  31         _ "nanomsg.org/go/mangos/v2/transport/all"
  32         "routing-manager/pkg/rtmgr"
  33         "strconv"
  34 )
  35
  36 type NngPush struct {
  37         Sbi
  38         NewSocket CreateNewNngSocketHandler
  39 }
  40
  41 func NewNngPush() *NngPush {
  42         instance := new(NngPush)
  43         instance.NewSocket = createNewPushSocket
  44         return instance
  45 }
  46
  47 func createNewPushSocket() (NngSocket, error) {
  48         rtmgr.Logger.Debug("Invoked: createNewPushSocket()")
  49         socket, err := push.NewSocket()
  50         if err != nil {
  51                 return nil, errors.New("can't create new push socket due to:" + err.Error())
  52         }
  53         socket.SetPipeEventHook(pipeEventHandler)
  54         return socket, nil
  55 }
  56
  57 func pipeEventHandler(event mangos.PipeEvent, pipe mangos.Pipe) {
  58         rtmgr.Logger.Debug("Invoked: pipeEventHandler()")
  59         for _, ep := range rtmgr.Eps {
  60                 uri := DEFAULT_NNG_PIPELINE_SOCKET_PREFIX + ep.Ip + ":" + strconv.Itoa(DEFAULT_NNG_PIPELINE_SOCKET_NUMBER)
  61                 if uri == pipe.Address() {
  62                         switch event {
  63                         case 1:
  64                                 ep.IsReady = true
  65                                 rtmgr.Logger.Debug("Endpoint " + uri + " successfully attached")
  66                         default:
  67                                 ep.IsReady = false
  68                                 rtmgr.Logger.Debug("Endpoint " + uri + " has been detached")
  69                         }
  70                 }
  71         }
  72 }
  73
  74 func (c *NngPush) Initialize(ip string) error {
  75         return nil
  76 }
  77
  78 func (c *NngPush) Terminate() error {
  79         return nil
  80 }
  81
  82 func (c *NngPush) AddEndpoint(ep *rtmgr.Endpoint) error {
  83         var err error
  84         var socket NngSocket
  85         rtmgr.Logger.Debug("Invoked sbi.AddEndpoint")
  86         rtmgr.Logger.Debug("args: %v", (*ep))
  87         socket, err = c.NewSocket()
  88         if err != nil {
  89                 return errors.New("can't add new socket to endpoint:" + ep.Uuid + " due to: " + err.Error())
  90         }
  91         ep.Socket = socket
  92         err = c.dial(ep)
  93         if err != nil {
  94                 return errors.New("can't dial to endpoint:" + ep.Uuid + " due to: " + err.Error())
  95         }
  96         return nil
  97 }
  98
  99 func (c *NngPush) DeleteEndpoint(ep *rtmgr.Endpoint) error {
 100         rtmgr.Logger.Debug("Invoked sbi. DeleteEndpoint")
 101         rtmgr.Logger.Debug("args: %v", (*ep))
 102         if err:= ep.Socket.(NngSocket).Close(); err != nil {
 103                         return errors.New("can't close push socket of endpoint:" + ep.Uuid + " due to: " + err.Error())
 104                 }
 105         return nil
 106 }
 107
 108 func (c *NngPush) UpdateEndpoints(rcs *rtmgr.RicComponents) {
 109         c.updateEndpoints(rcs, c)
 110 }
 111
 112 /*
 113 NOTE: Asynchronous dial starts a goroutine which keep maintains the connection to the given endpoint
 114 */
 115 func (c *NngPush) dial(ep *rtmgr.Endpoint) error {
 116         rtmgr.Logger.Debug("Dialing to endpoint: " + ep.Uuid)
 117         uri := DEFAULT_NNG_PIPELINE_SOCKET_PREFIX + ep.Ip + ":" + strconv.Itoa(DEFAULT_NNG_PIPELINE_SOCKET_NUMBER)
 118         options := make(map[string]interface{})
 119         options[mangos.OptionDialAsynch] = true
 120         if err := ep.Socket.(NngSocket).DialOptions(uri, options); err != nil {
 121                 return errors.New("can't dial on push socket to " + uri + " due to: " + err.Error())
 122         }
 123         return nil
 124 }
 125
 126 func (c *NngPush) DistributeAll(policies *[]string) error {
 127         rtmgr.Logger.Debug("Invoked: sbi.DistributeAll")
 128         rtmgr.Logger.Debug("args: %v", (*policies))
 129         for _, ep := range rtmgr.Eps {
 130                 if ep.IsReady {
 131                         go c.send(ep, policies)
 132                 } else {
 133                         rtmgr.Logger.Warn("Endpoint " + ep.Uuid + " is not ready")
 134                 }
 135         }
 136         return nil
 137 }
 138
 139 func (c *NngPush) send(ep *rtmgr.Endpoint, policies *[]string) {
 140         rtmgr.Logger.Debug("Push policy to endpoint: "+ ep.Uuid)
 141         for _, pe := range *policies {
 142                 if err := ep.Socket.(NngSocket).Send([]byte(pe)); err != nil {
 143                         rtmgr.Logger.Error("Unable to send policy entry due to: " + err.Error())
 144                 }
 145         }
 146         rtmgr.Logger.Info("NNG PUSH to ednpoint " + ep.Uuid + ": OK (# of Entries:" + strconv.Itoa(len((*policies))) + ")")
 147 }