fix a bug that could cause "Helium.dispatch" to panic

discovery/helium/helium.go

@@ -78,6 +78,11 @@ func (h *Helium) start(ctx context.Context) {
 
 func (h *Helium) dispatch(status types.ServiceStatus) {
 	h.subs.Range(func(k, v interface{}) bool {
+		defer func() {
+			if err := recover(); err != nil {
+				log.Errorf(context.TODO(), "[dispatch] dispatch %s failed, err: %v", k, err)
+			}
+		}()
 		c, ok := v.(chan<- types.ServiceStatus)
 		if !ok {
 			log.Error("[WatchServiceStatus] failed to cast channel from map")

discovery/helium/helium_test.go

@@ -47,3 +47,33 @@ func TestHelium(t *testing.T) {
 	close(chAddr)
 	close(chStatus)
 }
+
+func TestPanic(t *testing.T) {
+	chAddr := make(chan []string)
+
+	store := &storemocks.Store{}
+	store.On("ServiceStatusStream", mock.Anything).Return(chAddr, nil)
+
+	grpcConfig := types.GRPCConfig{
+		ServiceDiscoveryPushInterval: time.Duration(1) * time.Second,
+	}
+	service := New(grpcConfig, store)
+
+	for i := 0; i < 1000; i++ {
+		go func() {
+			chStatus := make(chan types.ServiceStatus)
+			uuid := service.Subscribe(chStatus)
+			time.Sleep(time.Second)
+			service.Unsubscribe(uuid)
+			close(chStatus)
+		}()
+	}
+
+	go func() {
+		for i := 0; i < 1000; i++ {
+			chAddr <- []string{"hhh", "hhh2"}
+		}
+	}()
+
+	time.Sleep(5 * time.Second)
+}