Why Go
- Case Studies
  
  Common problems companies solve with Go
- Use Cases
  
  Stories about how and why companies use Go
- Security Policy
  
  How Go can help keep you secure by default
Learn
Docs
- Effective Go
  
  Tips for writing clear, performant, and idiomatic Go code
- Go User Manual
  
  A complete introduction to building software with Go
- Standard library
  
  Reference documentation for Go's standard library
- Release Notes
  
  Learn what's new in each Go release
Packages
Community
- Recorded Talks
  
  Videos from prior events
- Meetups
  
  Meet other local Go developers
- Conferences
  
  Learn and network with Go developers from around the world
- Go blog
  
  The Go project's official blog.
- Go project
  
  Get help and stay informed from Go
- Get connected

redis

package module

Go to main page

Versions in this module

v0

v0.3.0

Apr 30, 2024

Changes in this version

type Consumer

+ func (c *Consumer) Pause(streams ...string) error

+ func (c *Consumer) Resume(streams ...string) error

v0.2.1

Sep 5, 2023

Changes in this version

+ type DecodeMessageContentOption interface

+ func WithMessageStateKeyPrefix(prefix string) DecodeMessageContentOption

+ type DecodeMessageContentOptionFunc func(setting *DecodeMessageContentSetting)

+ type DecodeMessageContentSetting struct

+ MessageStateKeyPrefix string

v0.2.0

Jun 30, 2023

Changes in this version

+ const LOGGER_PREFIX

+ const MAX_PENDING_FETCHING_SIZE

+ const MESSAGE_STATE_NAME_MAX_LENGTH

+ const MESSAGE_STATE_VALUE_MAX_SIZE

+ const MIN_PENDING_FETCHING_SIZE

+ const Nil

+ const PENDING_FETCHING_SIZE_COEFFICIENT

+ const StreamAsteriskID

+ const StreamLastDeliveredID

+ const StreamNeverDeliveredOffset

+ const StreamUnspecifiedOffset

+ const StreamZeroID

+ const StreamZeroOffset

+ func DefaultLogger() *log.Logger

+ type AdminClient struct

+ func NewAdminClient(opt *UniversalOptions) (*AdminClient, error)

+ func (c *AdminClient) Close() error

+ func (c *AdminClient) CreateConsumerGroup(stream, group, offset string) (string, error)

+ func (c *AdminClient) CreateConsumerGroupAndStream(stream, group, offset string) (string, error)

+ func (c *AdminClient) DeleteConsumer(stream, group, consumer string) (int64, error)

+ func (c *AdminClient) DeleteConsumerGroup(stream, group string) (int64, error)

+ func (c *AdminClient) Handle() redis.UniversalClient

+ func (c *AdminClient) SetConsumerGroupOffset(stream, group, offset string) (string, error)

+ type Consumer struct

+ ClaimMinIdleTime time.Duration

+ ClaimOccurrenceRate int32

+ ClaimSensitivity int

+ Group string

+ IdlingTimeout time.Duration

+ Logger *log.Logger

+ MaxInFlight int64

+ MaxPollingTimeout time.Duration

+ MessageHandler MessageHandleProc

+ Name string

+ RedisErrorHandler RedisErrorHandleProc

+ RedisOption *redis.UniversalOptions

+ func (c *Consumer) Close()

+ func (c *Consumer) Subscribe(streams ...StreamOffsetInfo) error

+ type ConsumerClient struct

+ Group string

+ Name string

+ RedisOption *redis.UniversalOptions

+ type ConsumerError struct

+ func (e *ConsumerError) Error() string

+ func (e *ConsumerError) IsRedisError() bool

+ func (e *ConsumerError) Unwrap() error

+ type CyclicCounter struct

+ type Forwarder struct

+ func NewForwarder(config *ProducerConfig) (*Forwarder, error)

+ func (f *Forwarder) Runner() *ForwarderRunner

+ type ForwarderRunner struct

+ func (r *ForwarderRunner) Start()

+ func (r *ForwarderRunner) Stop()

+ type Message struct

+ ConsumerGroup string

+ Delegate MessageDelegate

+ Stream string

+ func (m *Message) Ack()

+ func (m *Message) Clone() *Message

+ func (m *Message) Content() *MessageContent

+ func (m *Message) Del()

+ func (m *Message) HasResponded() bool

+ type MessageContent struct

+ State MessageState

+ Values map[string]interface{}

+ func DecodeMessageContent(container map[string]interface{}) *MessageContent

+ func NewMessageContent() *MessageContent

+ func (c *MessageContent) WriteTo(container map[string]interface{})

+ type MessageDelegate interface

+ OnAck func(msg *Message)

+ OnDel func(msg *Message)

+ type MessageHandleProc func(message *Message)

+ type MessageState struct

+ func (s *MessageState) Del(name string) interface{}

+ func (s *MessageState) Has(name string) bool

+ func (s *MessageState) Len() int

+ func (s *MessageState) Set(name string, value interface{}) (old interface{}, err error)

+ func (s *MessageState) SetString(name string, value interface{}) (old interface{}, err error)

+ func (s *MessageState) Value(name string) interface{}

+ func (s *MessageState) Visit(visit func(name string, value interface{}))

+ type ProduceMessageContentOption func(msg *MessageContent) error

+ func WithTracePropagation(ctx context.Context, propagator propagation.TextMapPropagator) ProduceMessageContentOption

+ type ProduceMessageIDOption func(id string) string

+ func WithMessageID(id string) ProduceMessageIDOption

+ type ProduceMessageOption interface

+ type Producer struct

+ func NewProducer(config *ProducerConfig) (*Producer, error)

+ func (p *Producer) Close()

+ func (p *Producer) Handle() redis.UniversalClient

+ func (p *Producer) Write(stream string, values map[string]interface{}, opts ...ProduceMessageOption) (string, error)

+ func (p *Producer) WriteContent(stream string, msg *MessageContent, opts ...ProduceMessageOption) (string, error)

+ type ProducerConfig struct

+ Logger *log.Logger

+ type RedisError interface

+ RedisError func()

+ type RedisErrorHandleProc func(err error) (disposed bool)

+ type Stream string

+ func (s Stream) NeverDeliveredOffset() StreamOffset

+ func (s Stream) Offset(offset string) StreamOffset

+ func (s Stream) Zero() StreamOffset

+ type StreamOffset struct

+ Offset string

+ Stream string

+ type StreamOffsetInfo interface

+ type UniversalClient = redis.UniversalClient

+ type UniversalOptions = redis.UniversalOptions

+ type XMessage = redis.XMessage

+ type XStream = redis.XStream

v0.2.0-alpha.20230629175308

Jun 29, 2023

v0.1.0

Jun 23, 2022