diff options
Diffstat (limited to 'rpc/server.go')
-rw-r--r-- | rpc/server.go | 17 |
1 files changed, 8 insertions, 9 deletions
diff --git a/rpc/server.go b/rpc/server.go index 78df37e52..62b84af34 100644 --- a/rpc/server.go +++ b/rpc/server.go @@ -21,6 +21,7 @@ import ( "fmt" "reflect" "runtime" + "strings" "sync" "sync/atomic" @@ -96,32 +97,30 @@ func (s *Server) RegisterName(name string, rcvr interface{}) error { return fmt.Errorf("%s is not exported", reflect.Indirect(rcvrVal).Type().Name()) } + methods, subscriptions := suitableCallbacks(rcvrVal, svc.typ) + // already a previous service register under given sname, merge methods/subscriptions if regsvc, present := s.services[name]; present { - methods, subscriptions := suitableCallbacks(rcvrVal, svc.typ) if len(methods) == 0 && len(subscriptions) == 0 { return fmt.Errorf("Service %T doesn't have any suitable methods/subscriptions to expose", rcvr) } - for _, m := range methods { regsvc.callbacks[formatName(m.method.Name)] = m } for _, s := range subscriptions { regsvc.subscriptions[formatName(s.method.Name)] = s } - return nil } svc.name = name - svc.callbacks, svc.subscriptions = suitableCallbacks(rcvrVal, svc.typ) + svc.callbacks, svc.subscriptions = methods, subscriptions if len(svc.callbacks) == 0 && len(svc.subscriptions) == 0 { return fmt.Errorf("Service %T doesn't have any suitable methods/subscriptions to expose", rcvr) } s.services[svc.name] = svc - return nil } @@ -303,7 +302,7 @@ func (s *Server) handle(ctx context.Context, codec ServerCodec, req *serverReque // active the subscription after the sub id was successfully sent to the client activateSub := func() { notifier, _ := NotifierFromContext(ctx) - notifier.activate(subid) + notifier.activate(subid, req.svcname) } return codec.CreateResponse(req.id, subid), activateSub @@ -383,7 +382,7 @@ func (s *Server) execBatch(ctx context.Context, codec ServerCodec, requests []*s codec.Close() } - // when request holds one of more subscribe requests this allows these subscriptions to be actived + // when request holds one of more subscribe requests this allows these subscriptions to be activated for _, c := range callbacks { c() } @@ -410,7 +409,7 @@ func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, Error) continue } - if r.isPubSub && r.method == unsubscribeMethod { + if r.isPubSub && strings.HasSuffix(r.method, unsubscribeMethodSuffix) { requests[i] = &serverRequest{id: r.id, isUnsubscribe: true} argTypes := []reflect.Type{reflect.TypeOf("")} // expect subscription id as first arg if args, err := codec.ParseRequestArguments(argTypes, r.params); err == nil { @@ -439,7 +438,7 @@ func (s *Server) readRequest(codec ServerCodec) ([]*serverRequest, bool, Error) } } } else { - requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{subscribeMethod, r.method}} + requests[i] = &serverRequest{id: r.id, err: &methodNotFoundError{r.method, r.method}} } continue } |