X-Git-Url: https://gerrit.o-ran-sc.org/r/gitweb?a=blobdiff_plain;f=pkg%2Fxapp%2Fsubscription.go;h=8f49e996867a179635c21efe98bc9e7f631852d8;hb=79f0680fd7bbf1c8a8c6e2a842cb18020e387a47;hp=27485dde76e1649b072439e2e8bb44fd08c3d3ca;hpb=47faec8c255f19f51e22c2fec203e6ed513c791f;p=ric-plt%2Fxapp-frame.git diff --git a/pkg/xapp/subscription.go b/pkg/xapp/subscription.go index 27485dd..8f49e99 100755 --- a/pkg/xapp/subscription.go +++ b/pkg/xapp/subscription.go @@ -23,34 +23,30 @@ import ( "bytes" "encoding/json" "fmt" + "io/ioutil" + "net/http" + "os" + "time" + "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" - "net/http" - "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, int) type SubscriptionQueryHandler func() (models.SubscriptionList, error) -type SubscriptionDeleteHandler func(string) error +type SubscriptionDeleteHandler func(string) int type SubscriptionResponseCallback func(*apimodel.SubscriptionResponse) type Subscriber struct { @@ -66,7 +62,11 @@ type Subscriber struct { func NewSubscriber(host string, timo int) *Subscriber { if host == "" { - host = "service-ricplt-submgr-http.ricplt:8088" + pltnamespace := os.Getenv("PLT_NAMESPACE") + if pltnamespace == "" { + pltnamespace = "ricplt" + } + host = fmt.Sprintf("service-%s-submgr-http.%s:8088", pltnamespace, pltnamespace) } if timo == 0 { @@ -109,39 +109,44 @@ func (r *Subscriber) Listen(createSubscription SubscriptionHandler, getSubscript api := operations.NewXappFrameworkAPI(swaggerSpec) // Subscription: Query - api.QueryGetAllSubscriptionsHandler = query.GetAllSubscriptionsHandlerFunc( - func(p query.GetAllSubscriptionsParams) middleware.Responder { + api.CommonGetAllSubscriptionsHandler = common.GetAllSubscriptionsHandlerFunc( + func(p common.GetAllSubscriptionsParams) middleware.Responder { if resp, err := getSubscription(); err == nil { - return query.NewGetAllSubscriptionsOK().WithPayload(resp) + return common.NewGetAllSubscriptionsOK().WithPayload(resp) } - return query.NewGetAllSubscriptionsInternalServerError() + return common.NewGetAllSubscriptionsInternalServerError() }) - // SubscriptionType: Report - api.ReportSubscribeReportHandler = report.SubscribeReportHandlerFunc( - func(p report.SubscribeReportParams) middleware.Responder { - if resp, err := createSubscription(models.SubscriptionTypeReport, p.ReportParams); err == nil { - return report.NewSubscribeReportCreated().WithPayload(resp) + // Subscription: Subscribe + api.CommonSubscribeHandler = common.SubscribeHandlerFunc( + func(params common.SubscribeParams) middleware.Responder { + resp, retCode := createSubscription(params.SubscriptionParams) + if retCode != common.SubscribeCreatedCode { + if retCode == common.SubscribeBadRequestCode { + return common.NewSubscribeBadRequest() + } else if retCode == common.SubscribeNotFoundCode { + return common.NewSubscribeNotFound() + } else if retCode == common.SubscribeServiceUnavailableCode { + return common.NewSubscribeServiceUnavailable() + } else { + return common.NewSubscribeInternalServerError() + } } - return report.NewSubscribeReportInternalServerError() + return common.NewSubscribeCreated().WithPayload(resp) }) - // SubscriptionType: Policy - api.PolicySubscribePolicyHandler = policy.SubscribePolicyHandlerFunc( - func(p policy.SubscribePolicyParams) middleware.Responder { - if resp, err := createSubscription(models.SubscriptionTypePolicy, p.PolicyParams); err == nil { - return policy.NewSubscribePolicyCreated().WithPayload(resp) - } - return policy.NewSubscribePolicyInternalServerError() - }) - - // SubscriptionType: Delete + // Subscription: Unsubscribe api.CommonUnsubscribeHandler = common.UnsubscribeHandlerFunc( func(p common.UnsubscribeParams) middleware.Responder { - if err := delSubscription(p.SubscriptionID); err == nil { - return common.NewUnsubscribeNoContent() + retCode := delSubscription(p.SubscriptionID) + if retCode != common.UnsubscribeNoContentCode { + if retCode == common.UnsubscribeBadRequestCode { + return common.NewUnsubscribeBadRequest() + } else { + return common.NewUnsubscribeInternalServerError() + } } - return common.NewUnsubscribeInternalServerError() + return common.NewUnsubscribeNoContent() }) server := restapi.NewServer(api) @@ -157,16 +162,14 @@ func (r *Subscriber) Listen(createSubscription SubscriptionHandler, getSubscript } // Server interface: send notification to client -func (r *Subscriber) Notify(resp *models.SubscriptionResponse, clientEndpoint string) (err error) { +func (r *Subscriber) Notify(resp *models.SubscriptionResponse, ep models.SubscriptionParamsClientEndpoint) (err error) { respData, err := json.Marshal(resp) if err != nil { Logger.Error("json.Marshal failed: %v", err) return err } - ep, _, _ := net.SplitHostPort(clientEndpoint) - _, port, _ := net.SplitHostPort(viper.GetString("local.host")) - clientUrl := fmt.Sprintf("http://%s:%s%s", ep, port, r.clientUrl) + clientUrl := fmt.Sprintf("http://%s:%d%s", ep.Host, *ep.HTTPPort, r.clientUrl) retries := viper.GetInt("subscription.retryCount") if retries == 0 { @@ -201,30 +204,18 @@ func (r *Subscriber) SetResponseCB(c SubscriptionResponseCallback) { r.clientCB = c } -// 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) +// 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 } - - return result.Payload, 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) - if err != nil { - return &apimodel.SubscriptionResponse{}, err - } - return result.Payload, err } // 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)