consumer

Jan 17, 2022

Changes in this version

type FirstToLastConsumerConfig

+ IPNSAddr string

Dec 11, 2021

Changes in this version

+ type Error error

+ var ErrContextDone Error = errors.New("context is done")

+ type MessageHandlerFunc func(ctx context.Context, message storage.ProtoUnmarshallable) error

+ func (m MessageHandlerFunc) Handle(ctx context.Context, message storage.ProtoUnmarshallable) error

Nov 26, 2021

Changes in this version

+ var ErrCommitNotFound = errors.New("commit not found")

+ type CommitReader interface

+ GetCommit func(ctx context.Context, cid cid.Cid) (commit, error)

+ type Consumer interface

+ Consume func(globalCtx context.Context, handler MessageHandler) error

+ type FakeCommitReader struct

+ Commits map[cid.Cid]commit

+ func (f FakeCommitReader) GetCommit(ctx context.Context, cid cid.Cid) (commit, error)

+ type FirstToLastConsumer struct

+ func NewFirstToLastConsumer(headReader thead.Reader, consumerOffsetManager thead.Manager, ...) *FirstToLastConsumer

+ func (f *FirstToLastConsumer) Consume(globalCtx context.Context, handler MessageHandler) error

+ type FirstToLastConsumerConfig struct

+ PollInterval time.Duration

+ PollTimeout time.Duration

+ type MessageFandlerFunc func(ctx context.Context, message storage.ProtoUnmarshallable) error

+ func (m MessageFandlerFunc) Handle(ctx context.Context, message storage.ProtoUnmarshallable) error

+ type MessageHandler interface

+ Handle func(ctx context.Context, message storage.ProtoUnmarshallable) error

+ type StorageCommitReader struct

+ func NewStorageCommitReader(reader storage.MessageReader, timeout time.Duration) *StorageCommitReader

+ func (cr *StorageCommitReader) GetCommit(ctx context.Context, cid cid.Cid) (commit, error)