package plugins import ( "fmt" "github.com/emqx/kuiper/common" "github.com/emqx/kuiper/xstream/api" "io/ioutil" "path" "sort" "strings" ) const ( baseProperty = `properties` baseOption = `options` ) type ( author struct { Name string `json:"name"` Email string `json:"email"` Company string `json:"company"` Website string `json:"website"` } language struct { English string `json:"en_US"` Chinese string `json:"zh_CN"` } field struct { Name string `json:"name"` Default interface{} `json:"default"` Type string `json:"type"` Control string `json:"control"` Optional bool `json:"optional"` Values interface{} `json:"values"` Hint *language `json:"hint"` Label *language `json:"label"` } sinkMeta struct { Author *author `json:"author"` HelpUrl *language `json:"helpUrl"` Libs []string `json:"libs"` Fields []*field `json:"properties"` } ) var g_sinkMetadata map[string]*sinkMeta //map[fileName] //const internal sinks var InternalSinks = [...]string{"log", "mqtt", "rest", "nop", "edgex"} func (this *Manager) readSinkMetaDir() error { confDir, err := common.GetConfLoc() if nil != err { return err } dir := path.Join(confDir, "sinks") tmpMap := make(map[string]*sinkMeta) //The internal support sinks for _, sink := range InternalSinks { file := path.Join(confDir, "sinks", "internal", sink + ".json") common.Log.Infof("Loading metadata file for sink: %s", file) meta := new(sinkMeta) err := common.ReadJsonUnmarshal(file, meta) if nil != err { return fmt.Errorf("Failed to load internal sink plugin:%s with err:%v", file, err) } tmpMap[sink + ".json"] = meta } files, err := ioutil.ReadDir(dir) if nil != err { return err } for _, file := range files { fname := file.Name() if !strings.HasSuffix(fname, ".json") { continue } filePath := path.Join(dir, fname) metadata := new(sinkMeta) err = common.ReadJsonUnmarshal(filePath, metadata) if nil != err { return fmt.Errorf("fname:%s err:%v", fname, err) } common.Log.Infof("sinkMeta file : %s", fname) tmpMap[fname] = metadata } g_sinkMetadata = tmpMap return nil } func (this *Manager) readSinkMetaFile(filePath string) error { metadata := new(sinkMeta) err := common.ReadJsonUnmarshal(filePath, metadata) if nil != err { return fmt.Errorf("filePath:%s err:%v", filePath, err) } sinkMetadata := g_sinkMetadata tmpMap := make(map[string]*sinkMeta) for k, v := range sinkMetadata { tmpMap[k] = v } fileName := path.Base(filePath) common.Log.Infof("sinkMeta file : %s", fileName) tmpMap[fileName] = metadata g_sinkMetadata = tmpMap return nil } type ( hintLanguage struct { English string `json:"en"` Chinese string `json:"zh"` } hintField struct { Name string `json:"name"` Default interface{} `json:"default"` Control string `json:"control"` Optional bool `json:"optional"` Type string `json:"type"` Hint *hintLanguage `json:"hint"` Label *hintLanguage `json:"label"` Values interface{} `json:"values"` } sinkPropertyNode struct { Fields []*hintField `json:"properties"` HelpUrl *hintLanguage `json:"helpUrl"` Libs []string `json:"libs"` } sinkProperty struct { CustomProperty map[string]*sinkPropertyNode `json:"customProperty"` BaseProperty map[string]*sinkPropertyNode `json:"baseProperty"` BaseOption *sinkPropertyNode `json:"baseOption"` } ) func (this *hintLanguage) set(l *language) { this.English = l.English this.Chinese = l.Chinese } func (this *hintField) setSinkField(v *field) { this.Name = v.Name this.Type = v.Type this.Default = v.Default this.Values = v.Values this.Control = v.Control this.Optional = v.Optional this.Hint = new(hintLanguage) this.Hint.set(v.Hint) this.Label = new(hintLanguage) this.Label.set(v.Label) } func (this *sinkPropertyNode) setNodeFromMetal(data *sinkMeta) { this.Libs = data.Libs if nil != data.HelpUrl { this.HelpUrl = new(hintLanguage) this.HelpUrl.set(data.HelpUrl) } for _, v := range data.Fields { field := new(hintField) field.setSinkField(v) this.Fields = append(this.Fields, field) } } func (this *sinkProperty) setCustomProperty(pluginName string) error { fileName := pluginName + `.json` sinkMetadata := g_sinkMetadata data := sinkMetadata[fileName] if nil == data { return fmt.Errorf(`not found pligin:%s`, fileName) } node := new(sinkPropertyNode) node.setNodeFromMetal(data) if 0 == len(this.CustomProperty) { this.CustomProperty = make(map[string]*sinkPropertyNode) } this.CustomProperty[pluginName] = node return nil } func (this *sinkProperty) setBasePropertry(pluginName string) error { sinkMetadata := g_sinkMetadata data := sinkMetadata[baseProperty+".json"] if nil == data { return fmt.Errorf(`not found pligin:%s`, baseProperty) } node := new(sinkPropertyNode) node.setNodeFromMetal(data) if 0 == len(this.BaseProperty) { this.BaseProperty = make(map[string]*sinkPropertyNode) } this.BaseProperty[pluginName] = node return nil } func (this *sinkProperty) setBaseOption() error { sinkMetadata := g_sinkMetadata data := sinkMetadata[baseOption+".json"] if nil == data { return fmt.Errorf(`not found pligin:%s`, baseOption) } node := new(sinkPropertyNode) node.setNodeFromMetal(data) this.BaseOption = node return nil } func (this *sinkProperty) hintWhenNewSink(pluginName string) (err error) { err = this.setCustomProperty(pluginName) if nil != err { return err } err = this.setBasePropertry(pluginName) if nil != err { return err } err = this.setBaseOption() return err } func (this *sinkPropertyNode) modifyPropertyNode(mapFields map[string]interface{}) (err error) { for i, field := range this.Fields { fieldVal := mapFields[field.Name] if nil != fieldVal { this.Fields[i].Default = fieldVal } } return nil } func (this *sinkProperty) modifyProperty(pluginName string, mapFields map[string]interface{}) (err error) { customProperty := this.CustomProperty[pluginName] if nil != customProperty { customProperty.modifyPropertyNode(mapFields) } baseProperty := this.BaseProperty[pluginName] if nil != baseProperty { baseProperty.modifyPropertyNode(mapFields) } return nil } func (this *sinkProperty) modifyOption(option *api.RuleOption) { baseOption := this.BaseOption if nil == baseOption { return } for i, field := range baseOption.Fields { switch field.Name { case `isEventTime`: baseOption.Fields[i].Default = option.IsEventTime case `lateTol`: baseOption.Fields[i].Default = option.LateTol case `concurrency`: baseOption.Fields[i].Default = option.Concurrency case `bufferLength`: baseOption.Fields[i].Default = option.BufferLength case `sendMetaToSink`: baseOption.Fields[i].Default = option.SendMetaToSink case `qos`: baseOption.Fields[i].Default = option.Qos case `checkpointInterval`: baseOption.Fields[i].Default = option.CheckpointInterval } } } func (this *sinkProperty) hintWhenModifySink(rule *api.Rule) (err error) { for _, m := range rule.Actions { for pluginName, sink := range m { mapFields, _ := sink.(map[string]interface{}) err = this.hintWhenNewSink(pluginName) if nil != err { return err } this.modifyProperty(pluginName, mapFields) } } this.modifyOption(rule.Options) return nil } func (this *Manager) GetSinkMeta(pluginName string, rule *api.Rule) (ptrSinkProperty *sinkProperty, err error) { ptrSinkProperty = new(sinkProperty) if nil == rule { err = ptrSinkProperty.hintWhenNewSink(pluginName) } else { err = ptrSinkProperty.hintWhenModifySink(rule) } return ptrSinkProperty, err } func (this *Manager) GetSinks() (sinks []string) { sinkMeta := g_sinkMetadata for fileName, _ := range sinkMeta { if fileName == baseProperty+".json" || fileName == baseOption+".json" { continue } sinks = append(sinks, strings.TrimSuffix(fileName, `.json`)) } sort.Sort(sort.StringSlice(sinks)) return sinks }