kafkaex

package

Go to main page

Versions in this module

v0

v0.0.6

May 7, 2024

Changes in this version

+ const APHMQH_EXECTYPE

type BoxMessage

+ func (m *BoxMessage) Blocked() bool

type Option

+ func WithExecType(execType int32) Option

type Options

+ ExecType int32

v0.0.5

May 7, 2024

Changes in this version

+ func ErrExec(topic, executer string, box *BoxMessage, err error, ...) error

+ func SetName(n string)

type IManager

+ RawPublish func(topic string, boxM *BoxMessage, ow func(*sarama.Config) *sarama.Config) error

type Option

+ func WithOverwrite(overwrite func(*sarama.Config) *sarama.Config) Option

type Options

+ Overwrite func(*sarama.Config) *sarama.Config

type WaterMillManager

+ func (m *WaterMillManager) RawPublish(topic string, boxM *BoxMessage, ow func(*sarama.Config) *sarama.Config) error

v0.0.4

May 6, 2024

v0.0.3

May 6, 2024

Changes in this version

+ const APHMQIGP_INNER

v0.0.2

May 6, 2024

v0.0.1

May 6, 2024

Changes in this version

+ const APHMQH_EXECAT

+ const APHMQH_EXECER

+ const APHMQH_EXECERR

+ const APHMQH_EXEC_TIMEOUT

+ const APHMQH_MSG_GROUP

+ const APHMQH_MSG_ID

+ const APHMQH_MSG_TOPIC

+ const APHMQH_PARTITION_KEY

+ const APHMQH_RETRIES

+ const APHMQH_RETRIES_MAX

+ const APHMQH_TRACE_ID

+ const APHMQIGP_DEF

+ const APHMQITP_DEAD

+ const APHMQITP_RETRY

+ var ErrNoFoundGroup = errors.New("没有配置组名")

+ var ErrNoFoundHandle = errors.New("没有配置执行函数")

+ var ErrNoFoundManager = errors.New("没有配置管理器")

+ var ErrNoFoundPublisher = errors.New("没有配置发布者")

+ var ErrNoFoundSubscriber = errors.New("没有配置订阅者")

+ var ErrNoFoundTopic = errors.New("没有配置主题")

+ func Field2ZapField(fs watermill.LogFields) []zapcore.Field

+ func NewPublisher(topic string) (*kafka.Publisher, error)

+ func NewSubscriber(group string) (*kafka.Subscriber, error)

+ func SetGetKafkaBrokersFunc(f func() []string)

+ func SetGetKafkaPwdFunc(f func() string)

+ func SetGetKafkaUserFunc(f func() string)

+ func SetRetryDelay(delay time.Duration)

+ func SetTimeout(timeout time.Duration)

+ func WithTrace(ctx context.Context) []zapcore.Field

+ type BoxMessage struct

+ ExecAt int64

+ ExecErr string

+ Execer string

+ MsgId string

+ Value []byte

+ func NewBoxMessage() *BoxMessage

+ func (m *BoxMessage) Dead() bool

+ func (m *BoxMessage) ExecResult(execer string, err error)

+ func (m *BoxMessage) NewRawMessage() *message.Message

+ func (m *BoxMessage) WithHeadersOption(headers map[string]string) *BoxMessage

+ func (m *BoxMessage) WithOption(opts ...Option) *BoxMessage

+ func (m *BoxMessage) WithRawMessage(msg *message.Message) *BoxMessage

+ type Handler func(ctx context.Context, box *BoxMessage) error

+ type ILogger interface

+ ErrorCtx func(ctx context.Context, msg string, args ...any)

+ InfoCtx func(ctx context.Context, msg string, args ...any)

+ type IManager interface

+ Publish func(topic string, boxmsg *BoxMessage) error

+ RegisterDead func(ctx context.Context, h Handler) error

+ RegisterRetry func(ctx context.Context, h Handler) error

+ RegisterSubscriber func(ctx context.Context, topic string, opts ...Option) error

+ func GetManager() IManager

+ func NewWaterMillManager() IManager

+ type Option func(*Options)

+ func WithGroup(group string) Option

+ func WithHandle(handle Handler) Option

+ func WithHandleTimeout(timeout time.Duration) Option

+ func WithKey(key string) Option

+ func WithRetryIndex(retryIndex int64) Option

+ func WithRetryMax(retryMax int64) Option

+ func WithSharedGroup() Option

+ func WithTopic(topic string) Option

+ func WithTraceID(traceId string) Option

+ type Options struct

+ Group string

+ Handle Handler

+ HandleTimeout time.Duration

+ Key string

+ RetryIndex int64

+ RetryMax int64

+ Topic string

+ TraceId string

+ func NewOptions(opts ...Option) *Options

+ func (o *Options) Fmt() *Options

+ func (o *Options) Verify() error

+ type WaterMillLogger struct

+ func NewWaterMillLogger() *WaterMillLogger

+ func (l *WaterMillLogger) Debug(msg string, fields watermill.LogFields)

+ func (l *WaterMillLogger) Error(msg string, err error, fields watermill.LogFields)

+ func (l *WaterMillLogger) ErrorCtx(ctx context.Context, msg string, args ...any)

+ func (l *WaterMillLogger) Info(msg string, fields watermill.LogFields)

+ func (l *WaterMillLogger) InfoCtx(ctx context.Context, msg string, args ...any)

+ func (l *WaterMillLogger) Trace(msg string, fields watermill.LogFields)

+ func (l *WaterMillLogger) With(fields watermill.LogFields) watermill.LoggerAdapter

+ type WaterMillManager struct

+ Pubs structure.ItemMap[kafka.Publisher]

+ Subs structure.ItemMap[kafka.Subscriber]

+ func (m *WaterMillManager) Publish(topic string, boxM *BoxMessage) error

+ func (m *WaterMillManager) RegisterDead(ctx context.Context, h Handler) error

+ func (m *WaterMillManager) RegisterRetry(ctx context.Context, h Handler) error

+ func (m *WaterMillManager) RegisterSubscriber(ctx context.Context, topic string, opts ...Option) error

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL