Execute wrappers before router

server/rpc_router.go

@@ -70,11 +70,9 @@ type router struct {
 	hdlrWrappers []HandlerWrapper
 }
 
-func newRpcRouter(opts Options) *router {
+func newRpcRouter() *router {
 	return &router{
-		name:         opts.Name,
+		serviceMap: make(map[string]*service),
-		hdlrWrappers: opts.HdlrWrappers,
-		serviceMap:   make(map[string]*service),
 	}
 }
 
@@ -207,10 +205,6 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
 			return nil
 		}
 
-		for i := len(router.hdlrWrappers); i > 0; i-- {
-			fn = router.hdlrWrappers[i-1](fn)
-		}
-
 		errmsg := ""
 		err := fn(ctx, r, replyv.Interface())
 		if err != nil {
@@ -252,10 +246,6 @@ func (s *service) call(ctx context.Context, router *router, sending *sync.Mutex,
 		}
 	}
 
-	for i := len(router.hdlrWrappers); i > 0; i-- {
-		fn = router.hdlrWrappers[i-1](fn)
-	}
-
 	// client.Stream request
 	r.stream = true
 

server/rpc_server.go

@@ -38,7 +38,7 @@ func newRpcServer(opts ...Option) Server {
 	options := newOptions(opts...)
 	return &rpcServer{
 		opts:        options,
-		router:      newRpcRouter(options),
+		router:      newRpcRouter(),
 		handlers:    make(map[string]Handler),
 		subscribers: make(map[*subscriber][]broker.Subscriber),
 		exit:        make(chan chan error),
@@ -133,16 +133,29 @@ func (s *rpcServer) ServeConn(sock transport.Socket) {
 
 		// set router
 		r := s.opts.Router
+
+		// if nil use default router
 		if s.opts.Router == nil {
 			r = s.router
 		}
 
-		// TODO: needs better error handling
+		// create a wrapped function
-		if err := r.ServeRequest(ctx, request, response); err != nil {
+		handler := func(ctx context.Context, req Request, rsp interface{}) error {
+			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 {
 			s.wg.Done()
 			log.Logf("Unexpected error serving request, closing socket: %v", err)
 			return
 		}
+
+		// done
 		s.wg.Done()
 	}
 }
@@ -171,7 +184,7 @@ func (s *rpcServer) Init(opts ...Option) error {
 	}
 
 	// update router
-	r := newRpcRouter(s.opts)
+	r := newRpcRouter()
 	r.serviceMap = s.router.serviceMap
 	s.router = r
 

server/server.go

@@ -122,7 +122,7 @@ var (
 	DefaultVersion        = "1.0.0"
 	DefaultId             = uuid.New().String()
 	DefaultServer  Server = newRpcServer()
-	DefaultRouter         = newRpcRouter(newOptions())
+	DefaultRouter         = newRpcRouter()
 )
 
 // DefaultOptions returns config options for the default service

server/subscriber.go

@@ -164,17 +164,29 @@ func validateSubscriber(sub Subscriber) error {
 func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handler {
 	return func(p broker.Publication) error {
 		msg := p.Message()
+
+		// get codec
 		ct := msg.Header["Content-Type"]
+
+		// default content type
+		if len(ct) == 0 {
+			msg.Header["Content-Type"] = DefaultContentType
+			ct = DefaultContentType
+		}
+
+		// get codec
 		cf, err := s.newCodec(ct)
 		if err != nil {
 			return err
 		}
 
+		// copy headers
 		hdr := make(map[string]string)
 		for k, v := range msg.Header {
 			hdr[k] = v
 		}
-		delete(hdr, "Content-Type")
+
+		// create context
 		ctx := metadata.NewContext(context.Background(), hdr)
 
 		results := make(chan error, len(sb.handlers))