go/go-micro
1
0
Fork 0
mirror of https://github.com/go-micro/go-micro.git synced 2024-12-24 10:07:04 +02:00
Code Issues Releases Activity

Some router changes

Browse Source
This commit is contained in:
Asim Aslam 2019-01-07 17:17:06 +00:00
parent 46ece968d4
commit c717af21ac
2 changed files with 28 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

42
server/rpc_router.go
View File

@ -166,28 +166,35 @@ func prepareMethod(method reflect.Method) *methodType {
return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream} return &methodType{method: method, ArgType: argType, ReplyType: replyType, ContextType: contextType, stream: stream}
} }
func (router *router) register(rcvr interface{}) error { func (router *router) NewHandler(h interface{}, opts ...HandlerOption) Handler {
return newRpcHandler(h, opts...)
}
func (router *router) Handle(h Handler) error {
router.mu.Lock() router.mu.Lock()
defer router.mu.Unlock() defer router.mu.Unlock()
if router.serviceMap == nil { if router.serviceMap == nil {
router.serviceMap = make(map[string]*service) router.serviceMap = make(map[string]*service)
} }
if len(h.Name()) == 0 {
return errors.New("rpc.Handle: handler has no name")
}
if !isExported(h.Name()) {
return errors.New("rpc.Handle: type " + h.Name() + " is not exported")
}
rcvr := h.Handler()
s := new(service) s := new(service)
s.typ = reflect.TypeOf(rcvr) s.typ = reflect.TypeOf(rcvr)
s.rcvr = reflect.ValueOf(rcvr) s.rcvr = reflect.ValueOf(rcvr)
sname := reflect.Indirect(s.rcvr).Type().Name()
if sname == "" { // check name
log.Fatal("rpc: no service name for type", s.typ.String()) if _, present := router.serviceMap[h.Name()]; present {
return errors.New("rpc.Handle: service already defined: " + h.Name())
} }
if !isExported(sname) {
s := "rpc Register: type " + sname + " is not exported" s.name = h.Name()
log.Log(s)
return errors.New(s)
}
if _, present := router.serviceMap[sname]; present {
return errors.New("rpc: service already defined: " + sname)
}
s.name = sname
s.method = make(map[string]*methodType) s.method = make(map[string]*methodType)
// Install the methods // Install the methods
@ -198,11 +205,12 @@ func (router *router) register(rcvr interface{}) error {
} }
} }
// Check there are methods
if len(s.method) == 0 { if len(s.method) == 0 {
s := "rpc Register: type " + sname + " has no exported methods of suitable type" return errors.New("rpc Register: type " + s.name + " has no exported methods of suitable type")
log.Log(s)
return errors.New(s)
} }
// save handler
router.serviceMap[s.name] = s router.serviceMap[s.name] = s
return nil return nil
} }
@ -318,7 +326,7 @@ func (m *methodType) prepareContext(ctx context.Context) reflect.Value {
return reflect.Zero(m.ContextType) return reflect.Zero(m.ContextType)
} }
func (router *router) serveRequest(ctx context.Context, codec serverCodec, ct string) error { func (router *router) ServeRequest(ctx context.Context, codec serverCodec, ct string) error {
sending := new(sync.Mutex) sending := new(sync.Mutex)
service, mtype, req, argv, replyv, keepReading, err := router.readRequest(codec) service, mtype, req, argv, replyv, keepReading, err := router.readRequest(codec)
if err != nil { if err != nil {

6
server/rpc_server.go
View File

@ -107,7 +107,7 @@ func (s *rpcServer) accept(sock transport.Socket) {
} }
// TODO: needs better error handling // TODO: needs better error handling
if err := s.router.serveRequest(ctx, codec, ct); err != nil { if err := s.router.ServeRequest(ctx, codec, ct); err != nil {
s.wg.Done() s.wg.Done()
log.Logf("Unexpected error serving request, closing socket: %v", err) log.Logf("Unexpected error serving request, closing socket: %v", err)
return return
@ -149,14 +149,14 @@ func (s *rpcServer) Init(opts ...Option) error {
} }
func (s *rpcServer) NewHandler(h interface{}, opts ...HandlerOption) Handler { func (s *rpcServer) NewHandler(h interface{}, opts ...HandlerOption) Handler {
return newRpcHandler(h, opts...) return s.router.NewHandler(h, opts...)
} }
func (s *rpcServer) Handle(h Handler) error { func (s *rpcServer) Handle(h Handler) error {
s.Lock() s.Lock()
defer s.Unlock() defer s.Unlock()
if err := s.router.register(h.Handler()); err != nil { if err := s.router.Handle(h); err != nil {
return err return err
} }