fix(connectino): share connection do not take effect

Signed-off-by: Jianxiang Ran <rxan_embedded@163.com>

internal/conf/connect_selector.go

@@ -25,29 +25,21 @@ func (c *ConSelector) Init() error {
 
 func (c *ConSelector) ReadCfgFromYaml() (props map[string]interface{}, err error) {
 
-	var (
-		found = false
-	)
-
-	cfg := make(map[string]interface{})
-	err = LoadConfigByName(CONNECTION_CONF, &cfg)
+	yamlOps, err := NewConfigOperatorFromConnectionYaml(c.Type)
 	if err != nil {
 		return nil, err
 	}
 
-	if cons, ok := cfg[c.Type]; ok {
-		if connItems, ok1 := cons.(map[string]interface{}); ok1 {
-			if conItem, ok := connItems[c.CfgKey]; ok {
-				if item, ok1 := conItem.(map[string]interface{}); ok1 {
-					props = item
-					found = true
-				}
-			}
+	cfg := yamlOps.CopyConfContent()
+	if len(cfg) == 0 {
+		return nil, fmt.Errorf("fail to parse yaml for connection Type %s", c.Type)
+	} else {
+		if cons, found := cfg[c.CfgKey]; found {
+			props = cons
+		} else {
+			return nil, fmt.Errorf("not found connection Type and Selector:  %s.%s", c.Type, c.CfgKey)
 		}
 	}
-	if !found {
-		return nil, fmt.Errorf("not found connection Type and Selector:  %s.%s", c.Type, c.CfgKey)
-	}
 
 	jsonPath := "sources/" + c.Type + ".json"
 	if c.Type == "mqtt" {

internal/conf/yaml_config_ops.go

@@ -43,7 +43,6 @@ type ConfKeysOperator interface {
 //ConfigOperator define interface to query/add/update/delete the configs in disk
 type ConfigOperator interface {
 	ConfKeysOperator
-	IsSource() bool
 	SaveCfgToFile() error
 }
 
@@ -255,10 +254,6 @@ type SourceConfigKeysOps struct {
 	*ConfigKeys
 }
 
-func (c *SourceConfigKeysOps) IsSource() bool {
-	return true
-}
-
 func (c *SourceConfigKeysOps) SaveCfgToFile() error {
 	pluginName := c.pluginName
 	confDir, err := GetDataLoc()
@@ -281,10 +276,6 @@ type SinkConfigKeysOps struct {
 	*ConfigKeys
 }
 
-func (c *SinkConfigKeysOps) IsSource() bool {
-	return false
-}
-
 func (c *SinkConfigKeysOps) SaveCfgToFile() error {
 	pluginName := c.pluginName
 	confDir, err := GetDataLoc()
@@ -307,10 +298,6 @@ type ConnectionConfigKeysOps struct {
 	*ConfigKeys
 }
 
-func (p *ConnectionConfigKeysOps) IsSource() bool {
-	return false
-}
-
 func (p *ConnectionConfigKeysOps) SaveCfgToFile() error {
 	pluginName := p.pluginName
 	confDir, err := GetDataLoc()
@@ -379,7 +366,7 @@ func NewConfigOperatorFromSourceYaml(pluginName string) (ConfigOperator, error)
 	fileName = pluginName
 
 	filePath = path.Join(dir, fileName+`.yaml`)
-	_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
+	_ = LoadConfigFromPath(filePath, &c.dataCfg)
 
 	return c, nil
 }
@@ -415,7 +402,7 @@ func NewConfigOperatorFromSinkYaml(pluginName string) (ConfigOperator, error) {
 	dir := path.Join(dataDir, "sinks")
 
 	filePath := path.Join(dir, pluginName+`.yaml`)
-	_ = filex.ReadYamlUnmarshal(filePath, &c.dataCfg)
+	_ = LoadConfigFromPath(filePath, &c.dataCfg)
 
 	return c, nil
 }
@@ -450,7 +437,7 @@ func NewConfigOperatorFromConnectionYaml(pluginName string) (ConfigOperator, err
 	}
 	yamlPath := path.Join(confDir, "connections/connection.yaml")
 	yamlData := make(map[string]interface{})
-	err = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
+	err = LoadConfigFromPath(yamlPath, &yamlData)
 	if nil != err {
 		return nil, err
 	}
@@ -476,7 +463,7 @@ func NewConfigOperatorFromConnectionYaml(pluginName string) (ConfigOperator, err
 	}
 	yamlPath = path.Join(confDir, "connections/connection.yaml")
 	yamlData = make(map[string]interface{})
-	_ = filex.ReadYamlUnmarshal(yamlPath, &yamlData)
+	_ = LoadConfigFromPath(yamlPath, &yamlData)
 
 	if plgCnfs, ok := yamlData[pluginName]; ok {
 		if cf, ok1 := plgCnfs.(map[string]interface{}); ok1 {

internal/conf/yaml_config_ops_test.go

@@ -21,40 +21,25 @@ import (
 )
 
 func TestConfigKeys_LoadSourceFile(t *testing.T) {
-
-	mqttCfg, err := NewConfigOperatorFromSourceYaml("mqtt")
+	_, err := NewConfigOperatorFromSourceYaml("mqtt")
 	if err != nil {
 		t.Error(err)
 	}
-
-	expect := mqttCfg.IsSource()
-	if expect != true {
-		t.Error(expect)
-	}
 }
 
 func TestConfigKeys_LoadConnectionMqtt(t *testing.T) {
-	mqttCfg, err := NewConfigOperatorFromConnectionYaml("mqtt")
+	_, err := NewConfigOperatorFromConnectionYaml("mqtt")
 	if err != nil {
 		t.Error(err)
 	}
-
-	actual := mqttCfg.IsSource()
-	if actual != false {
-		t.Errorf("should be false, but actual is %v", actual)
-	}
 }
 
 func TestConfigKeys_LoadConnectionEdgex(t *testing.T) {
-	edgeXCfg, err := NewConfigOperatorFromConnectionYaml("edgex")
+	_, err := NewConfigOperatorFromConnectionYaml("edgex")
 	if err != nil {
 		t.Error(err)
 	}
 
-	actual := edgeXCfg.IsSource()
-	if actual != false {
-		t.Errorf("should be false, but actual is %v", actual)
-	}
 }
 
 func TestConfigKeys_Ops(t *testing.T) {

internal/meta/connectionMeta.go

@@ -16,10 +16,10 @@ package meta
 
 import (
 	"fmt"
+	"strings"
 )
 
 func GetConnectionMeta(connectionName, language string) (ptrSourceProperty *uiSource, err error) {
-
 	gSourcemetaLock.RLock()
 	defer gSourcemetaLock.RUnlock()
 
@@ -47,41 +47,40 @@ func GetConnectionPlugins() (sources []*pluginfo) {
 	ConfigManager.lock.RLock()
 	defer ConfigManager.lock.RUnlock()
 
-	for _, conf := range ConfigManager.cfgOperators {
-		if conf.IsSource() {
-			continue
-		}
+	for key, conf := range ConfigManager.cfgOperators {
+		if strings.HasPrefix(key, ConnectionCfgOperatorKeyPrefix) {
 
-		plugName := conf.GetPluginName()
+			plugName := conf.GetPluginName()
 
-		uiSourceRepKey := plugName + `.json`
-		gSourcemetaLock.RLock()
-		v, found := gSourcemetadata[uiSourceRepKey]
-		if !found {
+			uiSourceRepKey := plugName + `.json`
+			gSourcemetaLock.RLock()
+			v, found := gSourcemetadata[uiSourceRepKey]
+			if !found {
+				gSourcemetaLock.RUnlock()
+				continue
+			}
 			gSourcemetaLock.RUnlock()
-			continue
-		}
-		gSourcemetaLock.RUnlock()
 
-		node := new(pluginfo)
-		node.Name = plugName
+			node := new(pluginfo)
+			node.Name = plugName
 
-		if nil == v.About {
-			continue
-		}
-		node.About = v.About
-		i := 0
-		for ; i < len(sources); i++ {
-			if node.Name <= sources[i].Name {
+			if nil == v.About {
+				continue
+			}
+			node.About = v.About
+			i := 0
+			for ; i < len(sources); i++ {
+				if node.Name <= sources[i].Name {
+					sources = append(sources, node)
+					copy(sources[i+1:], sources[i:])
+					sources[i] = node
+					break
+				}
+			}
+			if len(sources) == i {
 				sources = append(sources, node)
-				copy(sources[i+1:], sources[i:])
-				sources[i] = node
-				break
 			}
 		}
-		if len(sources) == i {
-			sources = append(sources, node)
-		}
 	}
 	return sources
 }