聊聊storagetapper的pipe

codecraft · · 365 次点击 · · 开始浏览    
这是一个创建于 的文章,其中的信息可能已经有所发展或是发生改变。

本文主要研究一下storagetapper的pipe

Pipe

storagetapper/pipe/pipe.go

type Pipe interface {
    NewConsumer(topic string) (Consumer, error)
    NewProducer(topic string) (Producer, error)
    Type() string
    Config() *config.PipeConfig
    Close() error
}
Pipe接口定义了NewConsumer、NewProducer、Type、Config、Close方法

Consumer

storagetapper/pipe/pipe.go

type Consumer interface {
    Close() error
    //CloseOnFailure doesn't save offsets
    CloseOnFailure() error
    Message() chan interface{}
    Error() chan error
    FetchNext() (interface{}, error)
    //Allows to explicitly persists current consumer position
    SaveOffset() error

    //SetFormat allow to tell consumer the format of the file when there is no
    //header
    SetFormat(format string)
}
Consumer接口定义了Close、CloseOnFailure、Message、Error、FetchNext、SaveOffset、SetFormat方法

Producer

storagetapper/pipe/pipe.go

type Producer interface {
    Push(data interface{}) error
    PushK(key string, data interface{}) error
    PushSchema(key string, data []byte) error
    //PushBatch queues the messages instead of sending immediately
    PushBatch(key string, data interface{}) error
    //PushCommit writes out all the messages queued by PushBatch
    PushBatchCommit() error
    Close() error
    CloseOnFailure() error

    SetFormat(format string)

    PartitionKey(source string, key string) string
}
Producer接口定义了Push、PushK、PushSchema、PushBatch、PushBatchCommit、Close、CloseOnFailure、SetFormat、PartitionKey

Create

storagetapper/pipe/pipe.go

func Create(pipeType string, cfg *config.PipeConfig, db *sql.DB) (Pipe, error) {

    init := Pipes[strings.ToLower(pipeType)]
    if init == nil {
        return nil, fmt.Errorf("unsupported pipe: %s", strings.ToLower(pipeType))
    }

    pipe, err := init(cfg, db)
    if err != nil {
        return nil, err
    }

    return pipe, nil
}

type constructor func(cfg *config.PipeConfig, db *sql.DB) (Pipe, error)

//Pipes is the list of registered pipes
//Plugins insert their constructors into this map
var Pipes map[string]constructor

//registerPlugin should be called from plugin's init
func registerPlugin(name string, init constructor) {
    if Pipes == nil {
        Pipes = make(map[string]constructor)
    }
    Pipes[name] = init
}
Create方法根据pipeType、PipeConfig、db来创建pipe

小结

storagetapper的Pipe接口定义了NewConsumer、NewProducer、Type、Config、Close方法,其Create方法根据pipeType、PipeConfig、db来创建pipe。

doc


有疑问加站长微信联系(非本文作者)

本文来自:Segmentfault

感谢作者:codecraft

查看原文:聊聊storagetapper的pipe

入群交流(和以上内容无关):加入Go大咖交流群,或添加微信:liuxiaoyan-s 备注:入群;或加QQ群:692541889

365 次点击  
加入收藏 微博
暂无回复
添加一条新回复 (您需要 登录 后才能回复 没有账号 ?)
  • 请尽量让自己的回复能够对别人有帮助
  • 支持 Markdown 格式, **粗体**、~~删除线~~、`单行代码`
  • 支持 @ 本站用户;支持表情(输入 : 提示),见 Emoji cheat sheet
  • 图片支持拖拽、截图粘贴等方式上传