关于golang:聊聊storagetapper的pipe

本文次要钻研一下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

  • storagetapper

【腾讯云】轻量 2核2G4M,首年65元

阿里云限时活动-云数据库 RDS MySQL  1核2G配置 1.88/月 速抢

本文由乐趣区整理发布,转载请注明出处,谢谢。

您可能还喜欢...

发表回复

您的电子邮箱地址不会被公开。 必填项已用*标注

此站点使用Akismet来减少垃圾评论。了解我们如何处理您的评论数据