fix: allow subscribers to register and deregister multi times.
This commit is contained in:
parent
3f5cbf2bcd
commit
532edc786f
33
http.go
33
http.go
@ -9,12 +9,12 @@ import (
|
|||||||
"sort"
|
"sort"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
"github.com/micro/go-log"
|
log "github.com/micro/go-log"
|
||||||
"github.com/micro/go-micro/broker"
|
"github.com/micro/go-micro/broker"
|
||||||
"github.com/micro/go-micro/server"
|
|
||||||
"github.com/micro/go-micro/cmd"
|
"github.com/micro/go-micro/cmd"
|
||||||
"github.com/micro/go-micro/codec"
|
"github.com/micro/go-micro/codec"
|
||||||
"github.com/micro/go-micro/registry"
|
"github.com/micro/go-micro/registry"
|
||||||
|
"github.com/micro/go-micro/server"
|
||||||
"github.com/micro/go-plugins/codec/jsonrpc"
|
"github.com/micro/go-plugins/codec/jsonrpc"
|
||||||
"github.com/micro/go-plugins/codec/protorpc"
|
"github.com/micro/go-plugins/codec/protorpc"
|
||||||
)
|
)
|
||||||
@ -28,6 +28,7 @@ var (
|
|||||||
"application/octet-stream": protorpc.NewCodec,
|
"application/octet-stream": protorpc.NewCodec,
|
||||||
}
|
}
|
||||||
)
|
)
|
||||||
|
|
||||||
type httpServer struct {
|
type httpServer struct {
|
||||||
sync.Mutex
|
sync.Mutex
|
||||||
opts server.Options
|
opts server.Options
|
||||||
@ -35,6 +36,8 @@ type httpServer struct {
|
|||||||
exit chan chan error
|
exit chan chan error
|
||||||
registerOnce sync.Once
|
registerOnce sync.Once
|
||||||
subscribers map[*httpSubscriber][]broker.Subscriber
|
subscribers map[*httpSubscriber][]broker.Subscriber
|
||||||
|
// used for first registration
|
||||||
|
registered bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func init() {
|
func init() {
|
||||||
@ -162,6 +165,19 @@ func (h *httpServer) Register() error {
|
|||||||
|
|
||||||
h.registerOnce.Do(func() {
|
h.registerOnce.Do(func() {
|
||||||
log.Logf("Registering node: %s", opts.Name+"-"+opts.Id)
|
log.Logf("Registering node: %s", opts.Name+"-"+opts.Id)
|
||||||
|
})
|
||||||
|
|
||||||
|
if err := opts.Registry.Register(service, rOpts...); err != nil {
|
||||||
|
return err
|
||||||
|
}
|
||||||
|
|
||||||
|
h.Lock()
|
||||||
|
defer h.Unlock()
|
||||||
|
|
||||||
|
if h.registered {
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
h.registered = true
|
||||||
|
|
||||||
for sb, _ := range h.subscribers {
|
for sb, _ := range h.subscribers {
|
||||||
handler := h.createSubHandler(sb, opts)
|
handler := h.createSubHandler(sb, opts)
|
||||||
@ -171,14 +187,11 @@ func (h *httpServer) Register() error {
|
|||||||
}
|
}
|
||||||
sub, err := opts.Broker.Subscribe(sb.Topic(), handler, subOpts...)
|
sub, err := opts.Broker.Subscribe(sb.Topic(), handler, subOpts...)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
log.Logf("Registering subscriber: %s, err: %s", sb.Topic, err)
|
return err
|
||||||
return
|
|
||||||
}
|
}
|
||||||
h.subscribers[sb] = []broker.Subscriber{sub}
|
h.subscribers[sb] = []broker.Subscriber{sub}
|
||||||
}
|
}
|
||||||
})
|
return nil
|
||||||
|
|
||||||
return opts.Registry.Register(service, rOpts...)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
func (h *httpServer) Deregister() error {
|
func (h *httpServer) Deregister() error {
|
||||||
@ -194,6 +207,12 @@ func (h *httpServer) Deregister() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
h.Lock()
|
h.Lock()
|
||||||
|
if !h.registered {
|
||||||
|
h.Unlock()
|
||||||
|
return nil
|
||||||
|
}
|
||||||
|
h.registered = false
|
||||||
|
|
||||||
for sb, subs := range h.subscribers {
|
for sb, subs := range h.subscribers {
|
||||||
for _, sub := range subs {
|
for _, sub := range subs {
|
||||||
log.Logf("Unsubscribing from topic: %s", sub.Topic())
|
log.Logf("Unsubscribing from topic: %s", sub.Topic())
|
||||||
|
Loading…
Reference in New Issue
Block a user