From 46d09ec2bcbd9c575dc9f498ae1cc7c899c9c54c Mon Sep 17 00:00:00 2001 From: Socket <30768657+Socketsj@users.noreply.github.com> Date: Thu, 30 Apr 2020 17:42:13 +0800 Subject: [PATCH] unsubscribe can async (#1596) Co-authored-by: huangshaojie Co-authored-by: Asim Aslam --- server/grpc/grpc.go | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/server/grpc/grpc.go b/server/grpc/grpc.go index 3de2332f..a4501c05 100644 --- a/server/grpc/grpc.go +++ b/server/grpc/grpc.go @@ -785,15 +785,21 @@ func (g *grpcServer) Deregister() error { g.registered = false + wg := sync.WaitGroup{} for sb, subs := range g.subscribers { for _, sub := range subs { - if logger.V(logger.InfoLevel, logger.DefaultLogger) { - logger.Infof("Unsubscribing from topic: %s", sub.Topic()) - } - sub.Unsubscribe() + wg.Add(1) + go func(s broker.Subscriber) { + defer wg.Done() + if logger.V(logger.InfoLevel, logger.DefaultLogger) { + logger.Infof("Unsubscribing from topic: %s", s.Topic()) + } + s.Unsubscribe() + }(sub) } g.subscribers[sb] = nil } + wg.Wait() g.Unlock() return nil