X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Fsubscription.go;h=6a6983ce4bd208a1cb6e0caf0d24ceb5a8c61643;hb=c79bd0e1a4fe855b06c8f8131f162d0ee4a5f9f5;hp=6ceabf98b802dc29e2787fb114ad1e229b98c4e7;hpb=b6341a5002c90da3d531a710f1038d716ef8c9bc;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/subscription.go b/pkg/xapp/subscription.go index 6ceabf9..6a6983c 100755 --- a/pkg/xapp/subscription.go +++ b/pkg/xapp/subscription.go @@ -20,34 +20,33 @@ package xapp import ( + "bytes" "encoding/json" "fmt" "github.com/go-openapi/loads" httptransport "github.com/go-openapi/runtime/client" "github.com/go-openapi/runtime/middleware" "github.com/go-openapi/strfmt" + "github.com/spf13/viper" "io/ioutil" "net/http" + "os" "time" apiclient "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi" apicommon "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi/common" - apipolicy "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi/policy" - apireport "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientapi/report" apimodel "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/clientmodel" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/models" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations" "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations/common" - "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations/policy" - "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations/query" - "gerrit.o-ran-sc.org/r/ric-plt/xapp-frame/pkg/restapi/operations/report" ) -type SubscriptionHandler func(models.SubscriptionType, interface{}) (*models.SubscriptionResponse, error) +type SubscriptionHandler func(interface{}) (*models.SubscriptionResponse, error) type SubscriptionQueryHandler func() (models.SubscriptionList, error) type SubscriptionDeleteHandler func(string) error +type SubscriptionResponseCallback func(*apimodel.SubscriptionResponse) type Subscriber struct { localAddr string @@ -56,29 +55,51 @@ type Subscriber struct { remoteUrl string remoteProt []string timeout time.Duration + clientUrl string + clientCB SubscriptionResponseCallback } func NewSubscriber(host string, timo int) *Subscriber { if host == "" { - host = "service-ricplt-submgr-http:8088" + pltnamespace := os.Getenv("PLT_NAMESPACE") + if pltnamespace == "" { + pltnamespace = "ricplt" + } + host = fmt.Sprintf("service-%s-submgr-http.%s:8088", pltnamespace, pltnamespace) } if timo == 0 { timo = 20 } - return &Subscriber{ + r := &Subscriber{ remoteHost: host, remoteUrl: "/ric/v1", remoteProt: []string{"http"}, timeout: time.Duration(timo) * time.Second, localAddr: "0.0.0.0", localPort: 8088, + clientUrl: "/ric/v1/subscriptions/response", + } + Resource.InjectRoute(r.clientUrl, r.ResponseHandler, "POST") + + return r +} + +func (r *Subscriber) ResponseHandler(w http.ResponseWriter, req *http.Request) { + if req.Body != nil { + var resp apimodel.SubscriptionResponse + if err := json.NewDecoder(req.Body).Decode(&resp); err == nil { + if r.clientCB != nil { + r.clientCB(&resp) + } + } + req.Body.Close() } } // Server interface: listen and receive subscription requests -func (r *Subscriber) Listen(add SubscriptionHandler, get SubscriptionQueryHandler, del SubscriptionDeleteHandler) error { +func (r *Subscriber) Listen(createSubscription SubscriptionHandler, getSubscription SubscriptionQueryHandler, delSubscription SubscriptionDeleteHandler) error { swaggerSpec, err := loads.Embedded(restapi.SwaggerJSON, restapi.FlatSwaggerJSON) if err != nil { return err @@ -86,37 +107,30 @@ func (r *Subscriber) Listen(add SubscriptionHandler, get SubscriptionQueryHandle api := operations.NewXappFrameworkAPI(swaggerSpec) - // Subscription: query - api.QueryGetAllSubscriptionsHandler = query.GetAllSubscriptionsHandlerFunc( - func(p query.GetAllSubscriptionsParams) middleware.Responder { - if resp, err := get(); err == nil { - return query.NewGetAllSubscriptionsOK().WithPayload(resp) - } - return query.NewGetAllSubscriptionsInternalServerError() - }) - - // SubscriptionType: Report - api.ReportSubscribeReportHandler = report.SubscribeReportHandlerFunc( - func(p report.SubscribeReportParams) middleware.Responder { - if resp, err := add(models.SubscriptionTypeReport, p.ReportParams); err == nil { - return report.NewSubscribeReportCreated().WithPayload(resp) + // Subscription: Query + api.CommonGetAllSubscriptionsHandler = common.GetAllSubscriptionsHandlerFunc( + func(p common.GetAllSubscriptionsParams) middleware.Responder { + if resp, err := getSubscription(); err == nil { + return common.NewGetAllSubscriptionsOK().WithPayload(resp) } - return report.NewSubscribeReportInternalServerError() + return common.NewGetAllSubscriptionsInternalServerError() }) - // SubscriptionType: policy - api.PolicySubscribePolicyHandler = policy.SubscribePolicyHandlerFunc( - func(p policy.SubscribePolicyParams) middleware.Responder { - if resp, err := add(models.SubscriptionTypePolicy, p.PolicyParams); err == nil { - return policy.NewSubscribePolicyCreated().WithPayload(resp) + // Subscription: Subscribe + api.CommonSubscribeHandler = common.SubscribeHandlerFunc( + func(params common.SubscribeParams) middleware.Responder { + Logger.Error("Subscribe: Params=%+v", params.SubscriptionParams) + if resp, err := createSubscription(params.SubscriptionParams); err == nil { + return common.NewSubscribeCreated().WithPayload(resp) } - return policy.NewSubscribePolicyInternalServerError() + return common.NewSubscribeInternalServerError() }) - // SubscriptionType: delete + // Subscription: Unsubscribe api.CommonUnsubscribeHandler = common.UnsubscribeHandlerFunc( func(p common.UnsubscribeParams) middleware.Responder { - if err := del(p.SubscriptionID); err == nil { + Logger.Error("Unsubscribe: SubscriptionID=%+v", p.SubscriptionID) + if err := delSubscription(p.SubscriptionID); err == nil { return common.NewUnsubscribeNoContent() } return common.NewUnsubscribeInternalServerError() @@ -134,21 +148,53 @@ func (r *Subscriber) Listen(add SubscriptionHandler, get SubscriptionQueryHandle return nil } -// Subscription interface for xApp: REPORT -func (r *Subscriber) SubscribeReport(p *apimodel.ReportParams) (*apimodel.SubscriptionResponse, error) { - params := apireport.NewSubscribeReportParamsWithTimeout(r.timeout).WithReportParams(p) - result, err := r.CreateTransport().Report.SubscribeReport(params) +// Server interface: send notification to client +func (r *Subscriber) Notify(resp *models.SubscriptionResponse, ep models.SubscriptionParamsClientEndpoint) (err error) { + respData, err := json.Marshal(resp) if err != nil { - return &apimodel.SubscriptionResponse{}, err + Logger.Error("json.Marshal failed: %v", err) + return err } - return result.Payload, err + clientUrl := fmt.Sprintf("http://%s:%d%s", ep.Host, *ep.HTTPPort, r.clientUrl) + + retries := viper.GetInt("subscription.retryCount") + if retries == 0 { + retries = 10 + } + + delay := viper.GetInt("subscription.retryDelay") + if delay == 0 { + delay = 5 + } + + for i := 0; i < retries; i++ { + r, err := http.Post(clientUrl, "application/json", bytes.NewBuffer(respData)) + if err == nil && (r != nil && r.StatusCode == http.StatusOK) { + break + } + + if err != nil { + Logger.Error("%v", err) + } + if r != nil && r.StatusCode != http.StatusOK { + Logger.Error("clientUrl=%s statusCode=%d", clientUrl, r.StatusCode) + } + time.Sleep(time.Duration(delay) * time.Second) + } + + return err } -// Subscription interface for xApp: POLICY -func (r *Subscriber) SubscribePolicy(p *apimodel.PolicyParams) (*apimodel.SubscriptionResponse, error) { - params := apipolicy.NewSubscribePolicyParamsWithTimeout(r.timeout).WithPolicyParams(p) - result, err := r.CreateTransport().Policy.SubscribePolicy(params) +// Subscription interface for xApp: Response callback +func (r *Subscriber) SetResponseCB(c SubscriptionResponseCallback) { + r.clientCB = c +} + +// Subscription interface for xApp +func (r *Subscriber) Subscribe(p *apimodel.SubscriptionParams) (*apimodel.SubscriptionResponse, error) { + params := apicommon.NewSubscribeParamsWithTimeout(r.timeout).WithSubscriptionParams(p) + result, err := r.CreateTransport().Common.Subscribe(params) if err != nil { return &apimodel.SubscriptionResponse{}, err } @@ -157,7 +203,7 @@ func (r *Subscriber) SubscribePolicy(p *apimodel.PolicyParams) (*apimodel.Subscr } // Subscription interface for xApp: DELETE -func (r *Subscriber) UnSubscribe(subId string) error { +func (r *Subscriber) Unsubscribe(subId string) error { params := apicommon.NewUnsubscribeParamsWithTimeout(r.timeout).WithSubscriptionID(subId) _, err := r.CreateTransport().Common.Unsubscribe(params) @@ -190,3 +236,32 @@ func (r *Subscriber) QuerySubscriptions() (models.SubscriptionList, error) { func (r *Subscriber) CreateTransport() *apiclient.RICSubscription { return apiclient.New(httptransport.New(r.remoteHost, r.remoteUrl, r.remoteProt), strfmt.Default) } + +/*func (r *Subscriber) getXappConfig() (appconfig models.XappConfigList, err error) { + + Logger.Error("Inside getXappConfig") + + var metadata models.ConfigMetadata + var xappconfig models.XAppConfig + name := viper.GetString("name") + configtype := "json" + metadata.XappName = &name + metadata.ConfigType = &configtype + + configFile, err := os.Open("/opt/ric/config/config-file.json") + if err != nil { + Logger.Error("Cannot open config file: %v", err) + return nil,errors.New("Could Not parse the config file") + } + + body, err := ioutil.ReadAll(configFile) + + defer configFile.Close() + + xappconfig.Metadata = &metadata + xappconfig.Config = body + + appconfig = append(appconfig,&xappconfig) + + return appconfig,nil +}*/