handle executes a request and returns the response from the callback.
(ctx context.Context, codec ServerCodec, req *serverRequest)
| 256 | |
| 257 | // handle executes a request and returns the response from the callback. |
| 258 | func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverRequest) (interface{}, func()) { |
| 259 | if req.err != nil { |
| 260 | return codec.CreateErrorResponse(&req.id, req.err), nil |
| 261 | } |
| 262 | |
| 263 | if req.isUnsubscribe { // cancel subscription, first param must be the subscription id |
| 264 | if len(req.args) >= 1 && req.args[0].Kind() == reflect.String { |
| 265 | notifier, supported := NotifierFromContext(ctx) |
| 266 | if !supported { // interface doesn't support subscriptions (e.g. http) |
| 267 | return codec.CreateErrorResponse(&req.id, &callbackError{ErrNotificationsUnsupported.Error()}), nil |
| 268 | } |
| 269 | |
| 270 | subid := ID(req.args[0].String()) |
| 271 | if err := notifier.unsubscribe(subid); err != nil { |
| 272 | return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil |
| 273 | } |
| 274 | |
| 275 | return codec.CreateResponse(req.id, true), nil |
| 276 | } |
| 277 | return codec.CreateErrorResponse(&req.id, &invalidParamsError{"Expected subscription id as first argument"}), nil |
| 278 | } |
| 279 | |
| 280 | if req.callb.isSubscribe { |
| 281 | subid, err := s.createSubscription(ctx, codec, req) |
| 282 | if err != nil { |
| 283 | return codec.CreateErrorResponse(&req.id, &callbackError{err.Error()}), nil |
| 284 | } |
| 285 | |
| 286 | // active the subscription after the sub id was successfully sent to the client |
| 287 | activateSub := func() { |
| 288 | notifier, _ := NotifierFromContext(ctx) |
| 289 | notifier.activate(subid, req.svcname) |
| 290 | } |
| 291 | |
| 292 | return codec.CreateResponse(req.id, subid), activateSub |
| 293 | } |
| 294 | |
| 295 | // regular RPC call, prepare arguments |
| 296 | if len(req.args) != len(req.callb.argTypes) { |
| 297 | rpcErr := &invalidParamsError{fmt.Sprintf("%s%s%s expects %d parameters, got %d", |
| 298 | req.svcname, serviceMethodSeparator, req.callb.method.Name, |
| 299 | len(req.callb.argTypes), len(req.args))} |
| 300 | return codec.CreateErrorResponse(&req.id, rpcErr), nil |
| 301 | } |
| 302 | |
| 303 | arguments := []reflect.Value{req.callb.rcvr} |
| 304 | if req.callb.hasCtx { |
| 305 | arguments = append(arguments, reflect.ValueOf(ctx)) |
| 306 | } |
| 307 | if len(req.args) > 0 { |
| 308 | arguments = append(arguments, req.args...) |
| 309 | } |
| 310 | |
| 311 | // execute RPC method and return result |
| 312 | reply := req.callb.method.Func.Call(arguments) |
| 313 | if len(reply) == 0 { |
| 314 | return codec.CreateResponse(req.id, nil), nil |
| 315 | } |
no test coverage detected