组件开发

推荐参考已有的组件开发一个新的组件

公共接口

  1. // 生命周期接口
  2. type Lifecycle interface {
  3.     // 初始化，例如初始化Kafka连接
  4.     Init(context Context)
  5.     // 启动运行，例如开始消费Kafka
  6.     Start()
  7.     // 停止
  8.     Stop()
  9. }
  11. // 描述接口
  12. type Describable interface {
  13.     // 类别，例如source
  14.     Category() Category
  15.     // 类型，例如kafka
  16.     Type() Type
  17.     // 自定义描述
  18.     String() string
  19. }
  21. // 配置获取接口
  22. type Config interface {
  23.     // 获取配置
  24.     Config() interface{}
  25. }
  27. // 组件接口
  28. type Component interface {
  29.     // 生命周期管理
  30.     Lifecycle
  31.     // 描述管理
  32.     Describable
  33.     // 配置管理
  34.     Config
  35. }

source组件

source组件对接数据源输入，开发一个新的source插件需要实现如下接口

  1. // source组件接口
  2. type Source interface {
  3.     Component
  4.     Producer
  5.     // 提交接口，确认sink端成功然后提交
  6.     Commit(events []Event)
  7. }
  9. // 生产接口，source组件需要实现
  10. type Producer interface {
  11.     // 对接数据源
  12.     ProductLoop(productFunc ProductFunc)
  13. }

sink组件

sink组件对接输出端，开发一个新的sink插件需要实现如下接口

  1.  // sink组件接口
  2. type Sink interface {
  3.     Component
  4.     Consumer
  5. }
  7. // 消费接口，sink组件需要实现
  8. type Consumer interface {
  9.     // 对接输出端
  10.     Consume(batch Batch) Result
  11. }

interceptor组件

interceptor组件对事件进行拦截处理，开发一个新的interceptor插件需要实现如下接口

  1. // interceptor组件接口
  2. type Interceptor interface {
  3.     Component
  4.     // 拦截处理
  5.     Intercept(invoker Invoker, invocation Invocation) api.Result
  6. }

Note

请注意新增的组件需要放到pkg/include/include.go的import当中注册