pkg/xapp/subscription.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 package xapp
  21
  22 import (
  23         "bytes"
  24         "encoding/json"
  25         "fmt"
  26         "github.com/go-openapi/loads"
  27         httptransport "github.com/go-openapi/runtime/client"
  28         "github.com/go-openapi/runtime/middleware"
  29         "github.com/go-openapi/strfmt"
  30         "github.com/spf13/viper"
  31         "io/ioutil"
  32         "net/http"
  33         "os"
  34         "time"
  35
  36         apiclient "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi"
  37         apicommon "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi/common"
  38         apimodel "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientmodel"
  39
  40         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models"
  41         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi"
  42         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations"
  43         "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations/common"
  44 )
  45
  46 type SubscriptionHandler func(interface{}) (*models.SubscriptionResponse, error)
  47 type SubscriptionQueryHandler func() (models.SubscriptionList, error)
  48 type SubscriptionDeleteHandler func(string) error
  49 type SubscriptionResponseCallback func(*apimodel.SubscriptionResponse)
  50
  51 type Subscriber struct {
  52         localAddr  string
  53         localPort  int
  54         remoteHost string
  55         remoteUrl  string
  56         remoteProt []string
  57         timeout    time.Duration
  58         clientUrl  string
  59         clientCB   SubscriptionResponseCallback
  60 }
  61
  62 func NewSubscriber(host string, timo int) *Subscriber {
  63         if host == "" {
  64                 pltnamespace := os.Getenv("PLT_NAMESPACE")
  65                 if pltnamespace == "" {
  66                         pltnamespace = "ricplt"
  67                 }
  68                 host = fmt.Sprintf("service-%s-submgr-http.%s:8088", pltnamespace, pltnamespace)
  69         }
  70
  71         if timo == 0 {
  72                 timo = 20
  73         }
  74
  75         r := &Subscriber{
  76                 remoteHost: host,
  77                 remoteUrl:  "/ric/v1",
  78                 remoteProt: []string{"http"},
  79                 timeout:    time.Duration(timo) * time.Second,
  80                 localAddr:  "0.0.0.0",
  81                 localPort:  8088,
  82                 clientUrl:  "/ric/v1/subscriptions/response",
  83         }
  84         Resource.InjectRoute(r.clientUrl, r.ResponseHandler, "POST")
  85
  86         return r
  87 }
  88
  89 func (r *Subscriber) ResponseHandler(w http.ResponseWriter, req *http.Request) {
  90         if req.Body != nil {
  91                 var resp apimodel.SubscriptionResponse
  92                 if err := json.NewDecoder(req.Body).Decode(&resp); err == nil {
  93                         if r.clientCB != nil {
  94                                 r.clientCB(&resp)
  95                         }
  96                 }
  97                 req.Body.Close()
  98         }
  99 }
 100
 101 // Server interface: listen and receive subscription requests
 102 func (r *Subscriber) Listen(createSubscription SubscriptionHandler, getSubscription SubscriptionQueryHandler, delSubscription SubscriptionDeleteHandler) error {
 103         swaggerSpec, err := loads.Embedded(restapi.SwaggerJSON, restapi.FlatSwaggerJSON)
 104         if err != nil {
 105                 return err
 106         }
 107
 108         api := operations.NewXappFrameworkAPI(swaggerSpec)
 109
 110         // Subscription: Query
 111         api.CommonGetAllSubscriptionsHandler = common.GetAllSubscriptionsHandlerFunc(
 112                 func(p common.GetAllSubscriptionsParams) middleware.Responder {
 113                         if resp, err := getSubscription(); err == nil {
 114                                 return common.NewGetAllSubscriptionsOK().WithPayload(resp)
 115                         }
 116                         return common.NewGetAllSubscriptionsInternalServerError()
 117                 })
 118
 119         // Subscription: Subscribe
 120         api.CommonSubscribeHandler = common.SubscribeHandlerFunc(
 121                 func(params common.SubscribeParams) middleware.Responder {
 122                         if resp, err := createSubscription(params.SubscriptionParams); err == nil {
 123                                 return common.NewSubscribeCreated().WithPayload(resp)
 124                         }
 125                         return common.NewSubscribeInternalServerError()
 126                 })
 127
 128         // Subscription: Unsubscribe
 129         api.CommonUnsubscribeHandler = common.UnsubscribeHandlerFunc(
 130                 func(p common.UnsubscribeParams) middleware.Responder {
 131                         if err := delSubscription(p.SubscriptionID); err == nil {
 132                                 return common.NewUnsubscribeNoContent()
 133                         }
 134                         return common.NewUnsubscribeInternalServerError()
 135                 })
 136
 137         server := restapi.NewServer(api)
 138         defer server.Shutdown()
 139         server.Host = r.localAddr
 140         server.Port = r.localPort
 141
 142         Logger.Info("Serving subscriptions on %s:%d\n", server.Host, server.Port)
 143         if err := server.Serve(); err != nil {
 144                 return err
 145         }
 146         return nil
 147 }
 148
 149 // Server interface: send notification to client
 150 func (r *Subscriber) Notify(resp *models.SubscriptionResponse, ep models.SubscriptionParamsClientEndpoint) (err error) {
 151         respData, err := json.Marshal(resp)
 152         if err != nil {
 153                 Logger.Error("json.Marshal failed: %v", err)
 154                 return err
 155         }
 156
 157         clientUrl := fmt.Sprintf("http://%s:%d%s", ep.Host, *ep.HTTPPort, r.clientUrl)
 158
 159         retries := viper.GetInt("subscription.retryCount")
 160         if retries == 0 {
 161                 retries = 10
 162         }
 163
 164         delay := viper.GetInt("subscription.retryDelay")
 165         if delay == 0 {
 166                 delay = 5
 167         }
 168
 169         for i := 0; i < retries; i++ {
 170                 r, err := http.Post(clientUrl, "application/json", bytes.NewBuffer(respData))
 171                 if err == nil && (r != nil && r.StatusCode == http.StatusOK) {
 172                         break
 173                 }
 174
 175                 if err != nil {
 176                         Logger.Error("%v", err)
 177                 }
 178                 if r != nil && r.StatusCode != http.StatusOK {
 179                         Logger.Error("clientUrl=%s statusCode=%d", clientUrl, r.StatusCode)
 180                 }
 181                 time.Sleep(time.Duration(delay) * time.Second)
 182         }
 183
 184         return err
 185 }
 186
 187 // Subscription interface for xApp: Response callback
 188 func (r *Subscriber) SetResponseCB(c SubscriptionResponseCallback) {
 189         r.clientCB = c
 190 }
 191
 192 // Subscription interface for xApp
 193 func (r *Subscriber) Subscribe(p *apimodel.SubscriptionParams) (*apimodel.SubscriptionResponse, error) {
 194         params := apicommon.NewSubscribeParamsWithTimeout(r.timeout).WithSubscriptionParams(p)
 195         result, err := r.CreateTransport().Common.Subscribe(params)
 196         if err != nil {
 197                 return &apimodel.SubscriptionResponse{}, err
 198         }
 199
 200         return result.Payload, err
 201 }
 202
 203 // Subscription interface for xApp: DELETE
 204 func (r *Subscriber) Unsubscribe(subId string) error {
 205         params := apicommon.NewUnsubscribeParamsWithTimeout(r.timeout).WithSubscriptionID(subId)
 206         _, err := r.CreateTransport().Common.Unsubscribe(params)
 207
 208         return err
 209 }
 210
 211 // Subscription interface for xApp: QUERY
 212 func (r *Subscriber) QuerySubscriptions() (models.SubscriptionList, error) {
 213         resp, err := http.Get(fmt.Sprintf("http://%s/%s/subscriptions", r.remoteHost, r.remoteUrl))
 214         if err != nil {
 215                 return models.SubscriptionList{}, err
 216         }
 217
 218         defer resp.Body.Close()
 219
 220         contents, err := ioutil.ReadAll(resp.Body)
 221         if err != nil {
 222                 return models.SubscriptionList{}, err
 223         }
 224
 225         subscriptions := models.SubscriptionList{}
 226         err = json.Unmarshal([]byte(string(contents)), &subscriptions)
 227         if err != nil {
 228                 return models.SubscriptionList{}, err
 229         }
 230
 231         return subscriptions, nil
 232 }
 233
 234 func (r *Subscriber) CreateTransport() *apiclient.RICSubscription {
 235         return apiclient.New(httptransport.New(r.remoteHost, r.remoteUrl, r.remoteProt), strfmt.Default)
 236 }
 237
 238 /*func (r *Subscriber) getXappConfig() (appconfig models.XappConfigList, err error) {
 239
 240     Logger.Error("Inside getXappConfig")
 241
 242                 var metadata models.ConfigMetadata
 243         var xappconfig models.XAppConfig
 244         name := viper.GetString("name")
 245         configtype := "json"
 246                 metadata.XappName = &name
 247                 metadata.ConfigType = &configtype
 248
 249         configFile, err := os.Open("/opt/ric/config/config-file.json")
 250         if err != nil {
 251                 Logger.Error("Cannot open config file: %v", err)
 252                 return nil,errors.New("Could Not parse the config file")
 253         }
 254
 255         body, err := ioutil.ReadAll(configFile)
 256
 257         defer configFile.Close()
 258
 259                 xappconfig.Metadata = &metadata
 260                 xappconfig.Config = body
 261
 262         appconfig = append(appconfig,&xappconfig)
 263
 264                 return appconfig,nil
 265 }*/