builtin.go 2.0 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667
  1. // Copyright 2021-2022 EMQ Technologies Co., Ltd.
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package io
  15. import (
  16. "github.com/lf-edge/ekuiper/internal/topo/memory"
  17. "github.com/lf-edge/ekuiper/internal/topo/neuron"
  18. "github.com/lf-edge/ekuiper/internal/topo/sink"
  19. "github.com/lf-edge/ekuiper/internal/topo/source"
  20. "github.com/lf-edge/ekuiper/pkg/api"
  21. )
  22. type NewSourceFunc func() api.Source
  23. type NewSinkFunc func() api.Sink
  24. var (
  25. sources = map[string]NewSourceFunc{
  26. "mqtt": func() api.Source { return &source.MQTTSource{} },
  27. "httppull": func() api.Source { return &source.HTTPPullSource{} },
  28. "file": func() api.Source { return &source.FileSource{} },
  29. "memory": func() api.Source { return memory.GetSource() },
  30. "neuron": func() api.Source { return neuron.GetSource() },
  31. }
  32. sinks = map[string]NewSinkFunc{
  33. "log": sink.NewLogSink,
  34. "logToMemory": sink.NewLogSinkToMemory,
  35. "mqtt": func() api.Sink { return &sink.MQTTSink{} },
  36. "rest": func() api.Sink { return &sink.RestSink{} },
  37. "nop": func() api.Sink { return &sink.NopSink{} },
  38. "memory": func() api.Sink { return memory.GetSink() },
  39. "neuron": func() api.Sink { return neuron.GetSink() },
  40. }
  41. )
  42. type Manager struct{}
  43. func (m *Manager) Source(name string) (api.Source, error) {
  44. if s, ok := sources[name]; ok {
  45. return s(), nil
  46. }
  47. return nil, nil
  48. }
  49. func (m *Manager) Sink(name string) (api.Sink, error) {
  50. if s, ok := sinks[name]; ok {
  51. return s(), nil
  52. }
  53. return nil, nil
  54. }
  55. var m = &Manager{}
  56. func GetManager() *Manager {
  57. return m
  58. }