goxstreams

Nov 20, 2023

Changes in this version

type Consumer

+ func (c Consumer[E]) WithConverter(convertTo func(event map[string]interface{}) (*E, error)) Consumer[E]

+ func (c Consumer[E]) WithCtxReceiver(...) Consumer[E]

type Producer

+ func (p Producer[E]) WithConverter(convertFrom func(event *E) (map[string]interface{}, error)) Producer[E]

+ func (p Producer[E]) WithCtxTransmmiter(transmitCtx func(ctx context.Context) (string, map[string]string)) Producer[E]

Oct 13, 2023

Changes in this version

type ConsumerConfig

+ ReadInterval time.Duration

type Producer

+ func (p Producer[E]) ProduceBatch(ctx context.Context, events []E, stream string) error

Sep 8, 2023

Changes in this version

type Consumer

+ func NewConsumerWithConverter(client RedisClient, worker Worker[E], ...) Consumer[E]

type Producer

+ func NewProducerWithConverter(client RedisClient, convertFrom func(event *E) (map[string]interface{}, error)) Producer[E]

Sep 1, 2023

Changes in this version

+ type RedisBrokenMessage struct

+ Body map[string]interface{}

+ Error error

+ ID string

+ RetryCount int

+ type RedisMessage struct

+ Body E

+ ID string

+ RetryCount int

Aug 31, 2023

Changes in this version

+ type Consumer struct

+ func NewConsumer(client RedisClient, converter ConsumerConverter[E], worker Worker[E], ...) Consumer[E]

+ func (c Consumer[E]) Run(ctx context.Context)

+ type ConsumerConfig struct

+ BatchSize int64

+ CleaneUp bool

+ ConsumerName string

+ FailIdle time.Duration

+ FailReadTime time.Duration

+ Group string

+ MaxConcurrency int64

+ MaxRetries int64

+ NoAck bool

+ Stream string

+ type ConsumerConverter interface

+ GetRedisID func(E) string

+ To func(id string, event map[string]interface{}) (E, error)

+ type Producer struct

+ func NewProducer(client RedisClient, converter ProducerConverter[E]) Producer[E]

+ func (p Producer[E]) Produce(ctx context.Context, event E, stream string) error

+ type ProducerConverter interface

+ From func(event E) map[string]interface{}

+ type RedisClient interface

+ type Worker interface

+ Process func(event E) error

+ ProcessBroken func(event map[string]interface{}) error

+ ProcessDead func(event E) error