Repository: lf-edge/ekuiper
Pull request #2904 (2024-06-05 03:26:10)

feat: revise error msg

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2904?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge) Report
All modified and coverable lines are covered by tests :white_check_mark:
> Please [upload](https://docs.codecov.com/docs/codecov-uploader) report for BASE (`master-v2@65a39f5`). [Learn more](https://docs.codecov.io/docs/error-reference?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge#section-missing-base-commit) about missing BASE report.


<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff              @@
##             master-v2    #2904   +/-   ##
============================================
  Coverage             ?   67.86%           
============================================
  Files                ?      294           
  Lines                ?    28192           
  Branches             ?        0           
============================================
  Hits                 ?    19130           
  Misses               ?     7361           
  Partials             ?     1701           
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/lf-edge/ekuiper/pull/2904?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=lf-edge).


Patch:
diff --git a/internal/binder/io/builtin.go b/internal/binder/io/builtin.go
index c021f983..fccf4f78 100644
--- a/internal/binder/io/builtin.go
+++ b/internal/binder/io/builtin.go
@@ -19,6 +19,7 @@ import (
 	"github.com/lf-edge/ekuiper/v2/internal/io/file"
 	"github.com/lf-edge/ekuiper/v2/internal/io/memory"
 	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt"
+	mqttCon "github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/io/sink"
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
@@ -45,6 +46,8 @@ func init() {
 
 	// modules.RegisterLookupSource("memory", func() api.LookupSource { return memory.GetLookupSource() })
 	// modules.RegisterLookupSource("httppull", func() api.LookupSource { return http.GetLookUpSource() })
+
+	modules.RegisterConnection("mqtt", mqttCon.CreateConnection)
 }
 
 type Manager struct{}
diff --git a/internal/io/connection/connection_pool.go b/internal/io/connection/connection_pool.go
index a2ff1ce3..3da1c89d 100644
--- a/internal/io/connection/connection_pool.go
+++ b/internal/io/connection/connection_pool.go
@@ -200,16 +200,21 @@ var globalConnectionManager *ConnectionManager
 
 func InitConnectionManager4Test() error {
 	InitMockTest()
-	return InitConnectionManager()
+	InitConnectionManager()
+	return nil
 }
 
-func InitConnectionManager() error {
+func InitConnectionManager() {
 	globalConnectionManager = &ConnectionManager{
 		connectionPool: make(map[string]ConnectionMeta),
 	}
 	if conf.IsTesting {
-		return nil
+		return
 	}
+	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
+}
+
+func ReloadConnection() error {
 	cfgs, err := conf.GetCfgFromKVStorage("connections", "", "")
 	if err != nil {
 		return err
@@ -228,12 +233,12 @@ func InitConnectionManager() error {
 		}
 		conn, err := createNamedConnection(context.Background(), meta)
 		if err != nil {
-			return fmt.Errorf("initialize connection:%v failed, err:%v", id, err)
+			conf.Log.Warnf("initialize connection:%v failed, err:%v", id, err)
+			continue
 		}
 		meta.conn = conn
 		globalConnectionManager.connectionPool[id] = meta
 	}
-	DefaultBackoffMaxElapsedDuration = time.Duration(conf.Config.Connection.BackoffMaxElapsedDuration)
 	return nil
 }
 
diff --git a/internal/io/mqtt/client/client.go b/internal/io/mqtt/client/client.go
index 92161fbf..fc006da1 100644
--- a/internal/io/mqtt/client/client.go
+++ b/internal/io/mqtt/client/client.go
@@ -32,10 +32,6 @@ import (
 	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
-func init() {
-	modules.ConnectionRegister["mqtt"] = CreateConnection
-}
-
 type Connection struct {
 	pahoMqtt.Client
 	selId     string
diff --git a/internal/pkg/filex/parser.go b/internal/pkg/filex/parser.go
index 4914508c..a3711c4e 100644
--- a/internal/pkg/filex/parser.go
+++ b/internal/pkg/filex/parser.go
@@ -16,6 +16,7 @@ package filex
 
 import (
 	"encoding/json"
+	"fmt"
 	"os"
 
 	"gopkg.in/yaml.v3"
@@ -41,7 +42,12 @@ func WriteYamlMarshal(path string, data interface{}) error {
 	return os.WriteFile(path, y, 0o666)
 }
 
-func ReadYamlUnmarshal(path string, ret interface{}) error {
+func ReadYamlUnmarshal(path string, ret interface{}) (err error) {
+	defer func() {
+		if err != nil {
+			err = fmt.Errorf("read %s unmarshal yaml failed, err:%v", path, err)
+		}
+	}()
 	sliByte, err := os.ReadFile(path)
 	if nil != err {
 		return err
diff --git a/internal/server/server.go b/internal/server/server.go
index cfaed7bd..b3930489 100644
--- a/internal/server/server.go
+++ b/internal/server/server.go
@@ -160,10 +160,6 @@ func StartUp(Version string) {
 	keyedstate.InitKeyedStateKV()
 
 	meta2.InitYamlConfigManager()
-	if err := connection.InitConnectionManager(); err != nil {
-		panic(err)
-	}
-
 	ruleProcessor = processor.NewRuleProcessor()
 	streamProcessor = processor.NewStreamProcessor()
 	rulesetProcessor = processor.NewRulesetProcessor(ruleProcessor, streamProcessor)
@@ -186,6 +182,10 @@ func StartUp(Version string) {
 	if err != nil {
 		panic(err)
 	}
+	connection.InitConnectionManager()
+	if err := connection.ReloadConnection(); err != nil {
+		conf.Log.Warn(err)
+	}
 	meta.Bind()
 	initRuleset()
 
diff --git a/pkg/modules/connection.go b/pkg/modules/connection.go
index feab1007..e693777b 100644
--- a/pkg/modules/connection.go
+++ b/pkg/modules/connection.go
@@ -25,10 +25,14 @@ type Connection interface {
 	Ref(ctx api.StreamContext) int
 }
 
-type RegisterConnection func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
+type ConnectionProvider func(ctx api.StreamContext, id string, props map[string]any) (Connection, error)
 
-var ConnectionRegister map[string]RegisterConnection
+var ConnectionRegister map[string]ConnectionProvider
 
 func init() {
-	ConnectionRegister = map[string]RegisterConnection{}
+	ConnectionRegister = map[string]ConnectionProvider{}
+}
+
+func RegisterConnection(name string, cp ConnectionProvider) {
+	ConnectionRegister[name] = cp
 }


Test patch:
diff --git a/internal/io/mqtt/source_test.go b/internal/io/mqtt/source_test.go
index 95495b15..0fefb880 100644
--- a/internal/io/mqtt/source_test.go
+++ b/internal/io/mqtt/source_test.go
@@ -23,9 +23,11 @@ import (
 
 	"github.com/lf-edge/ekuiper/v2/internal/conf"
 	"github.com/lf-edge/ekuiper/v2/internal/io/connection"
+	"github.com/lf-edge/ekuiper/v2/internal/io/mqtt/client"
 	"github.com/lf-edge/ekuiper/v2/internal/pkg/store"
 	"github.com/lf-edge/ekuiper/v2/internal/testx"
 	mockContext "github.com/lf-edge/ekuiper/v2/pkg/mock/context"
+	"github.com/lf-edge/ekuiper/v2/pkg/modules"
 )
 
 // NOTICE!!! Need to run a MQTT broker in localhost:1883 for this test or change the url to your broker
@@ -33,6 +35,7 @@ const url = "tcp://127.0.0.1:1883"
 
 func init() {
 	testx.InitEnv("mqtt_source_connector")
+	modules.RegisterConnection("mqtt", client.CreateConnection)
 }
 
 func TestProvision(t *testing.T) {
diff --git a/internal/pkg/filex/parser_test.go b/internal/pkg/filex/parser_test.go
new file mode 100644
index 00000000..c75a7981
--- /dev/null
+++ b/internal/pkg/filex/parser_test.go
@@ -0,0 +1,27 @@
+// Copyright 2024 EMQ Technologies Co., Ltd.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filex
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestReadYamlUnmarshalErr(t *testing.T) {
+	err := ReadYamlUnmarshal("1.yaml", map[string]interface{}{})
+	require.Error(t, err)
+	require.Equal(t, "read 1.yaml unmarshal yaml failed, err:open 1.yaml: no such file or directory", err.Error())
+}