mep/mepserver/mp1/controller.go

   1 /*
   2  * Copyright 2020 Huawei Technologies Co., Ltd.
   3  *
   4  * Licensed under the Apache License, Version 2.0 (the "License");
   5  * you may not use this file except in compliance with the License.
   6  * You may obtain a copy of the License at
   7  *
   8  *     http://www.apache.org/licenses/LICENSE-2.0
   9  *
  10  * Unless required by applicable law or agreed to in writing, software
  11  * distributed under the License is distributed on an "AS IS" BASIS,
  12  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13  * See the License for the specific language governing permissions and
  14  * limitations under the License.
  15  */
  16
  17 package mp1
  18
  19 import (
  20         "context"
  21         "net/http"
  22
  23         "github.com/apache/servicecomb-service-center/pkg/log"
  24         "github.com/apache/servicecomb-service-center/pkg/rest"
  25         "github.com/apache/servicecomb-service-center/pkg/util"
  26         "github.com/apache/servicecomb-service-center/server/core/proto"
  27         "github.com/apache/servicecomb-service-center/server/notify"
  28         v4 "github.com/apache/servicecomb-service-center/server/rest/controller/v4"
  29         svcutil "github.com/apache/servicecomb-service-center/server/service/util"
  30
  31         "mepserver/mp1/arch/workspace"
  32         "mepserver/mp1/models"
  33 )
  34
  35 const (
  36         basePath         = "/mep/mec_service_mgmt/v1"
  37         servicesPath     = basePath + "/services"
  38         appServicesPath  = basePath + "/applications/:appInstanceId" + "/services"
  39         appSubscribePath = basePath + "/applications/:appInstanceId/subscriptions"
  40 )
  41
  42 const (
  43         SerErrFailBase         workspace.ErrCode = workspace.TaskFail
  44         SerErrServiceNotFound                    = 2
  45         SerInstanceNotFound                      = 3
  46         ParseInfoErr                             = 4
  47         SubscriptionNotFound                     = 5
  48         OperateDataWithEtcdErr                   = 6
  49         SerErrServiceDelFailed                   = 7
  50         SerErrServiceUpdFailed                   = 8
  51 )
  52
  53 type APIHookFunc func() models.EndPointInfo
  54
  55 type APIGwHook struct {
  56         APIHook APIHookFunc
  57 }
  58
  59 var apihook APIGwHook
  60
  61 func SetAPIHook(hook APIGwHook) {
  62         apihook = hook
  63 }
  64
  65 func GetApiHook() APIGwHook {
  66         return apihook
  67 }
  68
  69 func init() {
  70         initRouter()
  71 }
  72
  73 func initRouter() {
  74         rest.
  75                 RegisterServant(&Mp1Service{})
  76 }
  77
  78 type Mp1Service struct {
  79         v4.MicroServiceService
  80 }
  81
  82 func (m *Mp1Service) URLPatterns() []rest.Route {
  83         return []rest.Route{
  84                 // appSubscriptions
  85                 {Method: rest.HTTP_METHOD_POST, Path: appSubscribePath, Func: doAppSubscribe},
  86                 {Method: rest.HTTP_METHOD_GET, Path: appSubscribePath, Func: getAppSubscribes},
  87                 {Method: rest.HTTP_METHOD_GET, Path: appSubscribePath + "/:subscriptionId", Func: getOneAppSubscribe},
  88                 {Method: rest.HTTP_METHOD_DELETE, Path: appSubscribePath + "/:subscriptionId", Func: delOneAppSubscribe},
  89                 // appServices
  90                 {Method: rest.HTTP_METHOD_POST, Path: appServicesPath, Func: serviceRegister},
  91                 {Method: rest.HTTP_METHOD_GET, Path: appServicesPath, Func: serviceDiscover},
  92                 {Method: rest.HTTP_METHOD_PUT, Path: appServicesPath + "/:serviceId", Func: serviceUpdate},
  93                 {Method: rest.HTTP_METHOD_GET, Path: appServicesPath + "/:serviceId", Func: getOneService},
  94                 {Method: rest.HTTP_METHOD_DELETE, Path: appServicesPath + "/:serviceId", Func: serviceDelete},
  95                 // services
  96                 {Method: rest.HTTP_METHOD_GET, Path: servicesPath, Func: serviceDiscover},
  97                 {Method: rest.HTTP_METHOD_GET, Path: servicesPath + "/:serviceId", Func: getOneService},
  98         }
  99 }
 100
 101 //application subscription
 102 func doAppSubscribe(w http.ResponseWriter, r *http.Request) {
 103
 104         workPlan := NewWorkSpace(w, r)
 105         workPlan.Try(
 106                 (&DecodeRestReq{}).WithBody(&models.SerAvailabilityNotificationSubscription{}),
 107                 &SubscribeIst{})
 108         workPlan.Finally(&SendHttpRspCreated{})
 109
 110         workspace.WkRun(workPlan)
 111 }
 112
 113 func getAppSubscribes(w http.ResponseWriter, r *http.Request) {
 114
 115         workPlan := NewWorkSpace(w, r)
 116         workPlan.Try(
 117                 &DecodeRestReq{},
 118                 &GetSubscribes{})
 119         workPlan.Finally(&SendHttpRsp{})
 120
 121         workspace.WkRun(workPlan)
 122 }
 123
 124 func getOneAppSubscribe(w http.ResponseWriter, r *http.Request) {
 125
 126         workPlan := NewWorkSpace(w, r)
 127         workPlan.Try(
 128                 &DecodeRestReq{},
 129                 &GetOneSubscribe{})
 130         workPlan.Finally(&SendHttpRsp{})
 131
 132         workspace.WkRun(workPlan)
 133 }
 134
 135 func delOneAppSubscribe(w http.ResponseWriter, r *http.Request) {
 136
 137         workPlan := NewWorkSpace(w, r)
 138         workPlan.Try(
 139                 &DecodeRestReq{},
 140                 &DelOneSubscribe{})
 141         workPlan.Finally(&SendHttpRsp{})
 142
 143         workspace.WkRun(workPlan)
 144 }
 145
 146 //service registery request
 147 func serviceRegister(w http.ResponseWriter, r *http.Request) {
 148         log.Info("Register service start...")
 149
 150         workPlan := NewWorkSpace(w, r)
 151         workPlan.Try(
 152                 (&DecodeRestReq{}).WithBody(&models.ServiceInfo{}),
 153                 &RegisterServiceId{},
 154                 &RegisterServiceInst{})
 155         workPlan.Finally(&SendHttpRspCreated{})
 156
 157         workspace.WkRun(workPlan)
 158 }
 159
 160 func serviceDiscover(w http.ResponseWriter, r *http.Request) {
 161         log.Info("Discover service service start...")
 162
 163         workPlan := NewWorkSpace(w, r)
 164         workPlan.Try(
 165                 &DiscoverDecode{},
 166                 &DiscoverService{},
 167                 &ToStrDiscover{},
 168                 &RspHook{})
 169         workPlan.Finally(&SendHttpRsp{})
 170
 171         workspace.WkRun(workPlan)
 172 }
 173
 174 func serviceUpdate(w http.ResponseWriter, r *http.Request) {
 175         log.Info("Update a service start...")
 176
 177         workPlan := NewWorkSpace(w, r)
 178         workPlan.Try(
 179                 (&DecodeRestReq{}).WithBody(&models.ServiceInfo{}),
 180                 &UpdateInstance{})
 181         workPlan.Finally(&SendHttpRsp{})
 182
 183         workspace.WkRun(workPlan)
 184 }
 185
 186 func getOneService(w http.ResponseWriter, r *http.Request) {
 187         log.Info("Register service start...")
 188
 189         workPlan := NewWorkSpace(w, r)
 190         workPlan.Try(
 191                 &GetOneDecode{},
 192                 &GetOneInstance{})
 193         workPlan.Finally(&SendHttpRsp{})
 194
 195         workspace.WkRun(workPlan)
 196
 197 }
 198
 199 func serviceDelete(w http.ResponseWriter, r *http.Request) {
 200         log.Info("Delete a service start...")
 201
 202         workPlan := NewWorkSpace(w, r)
 203         workPlan.Try(
 204                 &DecodeRestReq{},
 205                 &DeleteService{})
 206         workPlan.Finally(&SendHttpRsp{})
 207
 208         workspace.WkRun(workPlan)
 209 }
 210
 211 func WebsocketListAndWatch(ctx context.Context, req *proto.WatchInstanceRequest, consumerSvcId string) {
 212         if req == nil || len(req.SelfServiceId) == 0 {
 213                 log.Warn("request fomat invalid！")
 214                 return
 215         }
 216         domainProject := util.ParseDomainProject(ctx)
 217         if !svcutil.ServiceExist(ctx, domainProject, req.SelfServiceId) {
 218                 log.Warn("service does not exist!")
 219                 return
 220         }
 221         DoWebsocketListAndWatch(ctx, req.SelfServiceId, consumerSvcId, func() ([]*proto.WatchInstanceResponse, int64) {
 222                 return svcutil.QueryAllProvidersInstances(ctx, req.SelfServiceId)
 223         })
 224 }
 225
 226 func DoWebsocketListAndWatch(ctx context.Context, serviceId string, consumerSvcId string, f func() ([]*proto.WatchInstanceResponse, int64)) {
 227         domainProject := util.ParseDomainProject(ctx)
 228         socket := &Websocket{
 229                 ctx:       ctx,
 230                 watcher:   notify.NewInstanceEventListWatcher(serviceId, domainProject, f),
 231                 serviceID: consumerSvcId,
 232         }
 233         ProcessSocket(socket)
 234 }
 235
 236 func ProcessSocket(socket *Websocket) {
 237         if err := socket.Init(); err != nil {
 238                 return
 239         }
 240         socket.HandleWatchWebSocketControlMessage()
 241 }