ekuiper-new/internal/topo/node/conf/sink.go

// Copyright 2022-2023 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 conf

import (
	"github.com/lf-edge/ekuiper/internal/conf"
)

const ResourceID = "resourceId"

func GetSinkConf(sinkType string, action map[string]interface{}) map[string]interface{} {
	resourceId, ok := action[ResourceID].(string)
	if !ok {
		return action
	}
	delete(action, ResourceID)

	yamlOps, err := conf.NewConfigOperatorFromSinkYaml(sinkType)
	if err != nil {
		conf.Log.Warnf("fail to parse yaml for sink %s. Return error %v", sinkType, err)
		return action
	}
	var props map[string]interface{}
	cfg := yamlOps.CopyConfContent()
	if len(cfg) == 0 {
		conf.Log.Warnf("fail to parse yaml for sink %s. Return an empty configuration", sinkType)
		return action
	} else {
		def, ok := cfg[resourceId]
		if !ok {
			conf.Log.Warnf("resource id %s is not found", resourceId)
			return action
		} else {
			props = def
			for k, v := range action {
				props[k] = v
			}
		}
	}

	conf.Log.Debugf("get conf for %s with resource id %s: %v", sinkType, resourceId, printable(props))
	return props
}