add the wrappers back into the core router

This commit is contained in:
Asim Aslam 2019-06-07 15:02:19 +01:00
parent a2fbf19341
commit bfdec9e2e3
2 changed files with 47 additions and 15 deletions

View File

@ -202,6 +202,11 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
return nil return nil
} }
// wrap the handler
for i := len(router.hdlrWrappers); i > 0; i-- {
fn = router.hdlrWrappers[i-1](fn)
}
// execute handler // execute handler
if err := fn(ctx, r, replyv.Interface()); err != nil { if err := fn(ctx, r, replyv.Interface()); err != nil {
return err return err
@ -235,6 +240,11 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
} }
} }
// wrap the handler
for i := len(router.hdlrWrappers); i > 0; i-- {
fn = router.hdlrWrappers[i-1](fn)
}
// client.Stream request // client.Stream request
r.stream = true r.stream = true

View File

@ -35,6 +35,9 @@ type rpcServer struct {
func newRpcServer(opts ...Option) Server { func newRpcServer(opts ...Option) Server {
options := newOptions(opts...) options := newOptions(opts...)
router := newRpcRouter()
router.hdlrWrappers = options.HdlrWrappers
return &rpcServer{ return &rpcServer{
opts: options, opts: options,
router: DefaultRouter, router: DefaultRouter,
@ -45,6 +48,14 @@ func newRpcServer(opts ...Option) Server {
} }
} }
type rpcRouter struct {
h func(context.Context, Request, interface{}) error
}
func (r rpcRouter) ServeRequest(ctx context.Context, req Request, rsp Response) error {
return r.h(ctx, req, rsp)
}
// ServeConn serves a single connection // ServeConn serves a single connection
func (s *rpcServer) ServeConn(sock transport.Socket) { func (s *rpcServer) ServeConn(sock transport.Socket) {
defer func() { defer func() {
@ -143,24 +154,26 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
} }
// set router // set router
r := s.opts.Router r := Router(s.router)
// if nil use default router // if not nil use the router specified
if s.opts.Router == nil { if s.opts.Router != nil {
r = s.router // create a wrapped function
handler := func(ctx context.Context, req Request, rsp interface{}) error {
return s.opts.Router.ServeRequest(ctx, req, rsp.(Response))
}
// execute the wrapper for it
for i := len(s.opts.HdlrWrappers); i > 0; i-- {
handler = s.opts.HdlrWrappers[i-1](handler)
}
// set the router
r = rpcRouter{handler}
} }
// create a wrapped function // serve the actual request using the request router
handler := func(ctx context.Context, req Request, rsp interface{}) error { if err := r.ServeRequest(ctx, request, response); err != nil {
return r.ServeRequest(ctx, req, rsp.(Response))
}
for i := len(s.opts.HdlrWrappers); i > 0; i-- {
handler = s.opts.HdlrWrappers[i-1](handler)
}
// TODO: handle error better
if err := handler(ctx, request, response); err != nil {
// write an error response // write an error response
err = rcodec.Write(&codec.Message{ err = rcodec.Write(&codec.Message{
Header: msg.Header, Header: msg.Header,
@ -206,6 +219,15 @@ func (s *rpcServer) Init(opts ...Option) error {
for _, opt := range opts { for _, opt := range opts {
opt(&s.opts) opt(&s.opts)
} }
// update router if its the default
if s.opts.Router == nil {
r := newRpcRouter()
r.hdlrWrappers = s.opts.HdlrWrappers
r.serviceMap = s.router.serviceMap
s.router = r
}
s.Unlock() s.Unlock()
return nil return nil
} }