diff --git a/registry/kubernetes/registry.go b/registry/kubernetes/registry.go index 289a42aa5a859687b850cb9f8646f836691385ee..f8618caacca006ea1b8ab3940940bc93ea953f97 100644 --- a/registry/kubernetes/registry.go +++ b/registry/kubernetes/registry.go @@ -89,11 +89,12 @@ func (r *kubernetesRegistry) CloseAndNilClient() { func (r *kubernetesRegistry) CloseListener() { r.cltLock.Lock() - if r.configListener != nil { - r.configListener.Close() + l := r.configListener + r.cltLock.Unlock() + if l != nil { + l.Close() } r.configListener = nil - r.cltLock.Unlock() } func (r *kubernetesRegistry) CreatePath(k string) error { @@ -124,11 +125,11 @@ func (r *kubernetesRegistry) DoSubscribe(svc *common.URL) (registry.Listener, er return nil, perrors.New("kubernetes client broken") } - // new client & listener - listener := kubernetes.NewEventListener(r.client) - r.listenerLock.Lock() - r.listener = listener + if r.listener == nil { + // double check + r.listener = kubernetes.NewEventListener(r.client) + } r.listenerLock.Unlock() } diff --git a/remoting/kubernetes/client_test.go b/remoting/kubernetes/client_test.go index d8ed78291974c5e348d5edf72c077ab2a1f4021e..985aba6a94f7ccab15f8c1b92c0bcc3768b15a39 100644 --- a/remoting/kubernetes/client_test.go +++ b/remoting/kubernetes/client_test.go @@ -328,14 +328,16 @@ func (s *KubernetesClientTestSuite) TestClientGetChildrenKVList() { wg := sync.WaitGroup{} wg.Add(1) + syncDataComplete := make(chan struct{}) + go func() { - defer wg.Done() wc, done, err := client.WatchWithPrefix(prefix) if err != nil { t.Fatal(err) } i := 0 + wg.Done() for { select { @@ -344,6 +346,7 @@ func (s *KubernetesClientTestSuite) TestClientGetChildrenKVList() { t.Logf("got event %v k %s v %s", e.EventType, e.Key, e.Value) if i == 3 { // already sync all event + syncDataComplete <- struct{}{} return } case <-done: @@ -353,6 +356,9 @@ func (s *KubernetesClientTestSuite) TestClientGetChildrenKVList() { } }() + // wait the watch goroutine start + wg.Wait() + expect := make(map[string]string) got := make(map[string]string) @@ -370,9 +376,7 @@ func (s *KubernetesClientTestSuite) TestClientGetChildrenKVList() { } } - // must wait client sync all create event - - wg.Wait() + <-syncDataComplete // start get all children kList, vList, err := client.GetChildren(prefix)