queue

Documentation

Index

• func SCRAMClientGeneratorFuncSHA256() sarama.SCRAMClient
• func SCRAMClientGeneratorFuncSHA512() sarama.SCRAMClient
• type ConsumerGroupHandler
• type ConsumerRegister
• type Kafka
  • func NewKafka(brokers []string, c *sarama.Config, h ConsumerGroupHandler, provider string) (k *Kafka, err error)
  • func (e *Kafka) Append(opts ...storage.Option) error
  • func (e *Kafka) Register(opts ...storage.Option)
  • func (e *Kafka) Run(ctx context.Context)
  • func (e *Kafka) Shutdown()
  • func (*Kafka) String() string
• type KafkaRunReader
• type Memory
  • func NewMemory(poolNum uint) *Memory
  • func (m *Memory) Append(opts ...storage.Option) error
  • func (m *Memory) Register(opts ...storage.Option)
  • func (m *Memory) Run(context.Context)
  • func (m *Memory) Shutdown()
  • func (*Memory) String() string
• type Message
  • func (m *Message) GetContext() context.Context
  • func (m *Message) GetErrorCount() int
  • func (m *Message) GetID() string
  • func (m *Message) GetPrefix() (prefix string)
  • func (m *Message) GetStream() string
  • func (m *Message) GetValues() map[string]interface{}
  • func (m *Message) SetContext(ctx context.Context)
  • func (m *Message) SetErrorCount(count int)
  • func (m *Message) SetID(id string)
  • func (m *Message) SetPrefix(prefix string)
  • func (m *Message) SetStream(stream string)
  • func (m *Message) SetValues(values map[string]interface{})
• type MessageHandler
  • func (h *MessageHandler) Cleanup(sarama.ConsumerGroupSession) error
  • func (h *MessageHandler) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error
  • func (h *MessageHandler) SetConsumerFunc(f storage.ConsumerFunc)
  • func (h *MessageHandler) Setup(s sarama.ConsumerGroupSession) error
• type NSQ
  • func NewNSQ(cfg *nsq.Config, lookup, adminAddr string, addresses ...string) (*NSQ, error)
  • func (e *NSQ) Append(opts ...storage.Option) error
  • func (e *NSQ) Register(opts ...storage.Option)
  • func (e *NSQ) Run(context.Context)
  • func (e *NSQ) Shutdown()
  • func (*NSQ) String() string
• type NodesResp
• type Redis
  • func NewRedis(producerOptions *redisqueue.ProducerOptions, ...) (*Redis, error)
  • func (r *Redis) Append(opts ...storage.Option) error
  • func (r *Redis) Register(opts ...storage.Option)
  • func (r *Redis) Run(context.Context)
  • func (r *Redis) Shutdown()
  • func (*Redis) String() string
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Functions

func SCRAMClientGeneratorFuncSHA256

func SCRAMClientGeneratorFuncSHA256() sarama.SCRAMClient

func SCRAMClientGeneratorFuncSHA512

func SCRAMClientGeneratorFuncSHA512() sarama.SCRAMClient

Types

type ConsumerGroupHandler

type ConsumerGroupHandler interface {
	sarama.ConsumerGroupHandler
	SetConsumerFunc(f storage.ConsumerFunc)
}

type ConsumerRegister

type ConsumerRegister struct {
	Topic     string
	GroupID   string
	Partition int
	Func      ConsumerGroupHandler
}

type Kafka

type Kafka struct {
	// contains filtered or unexported fields
}

func NewKafka

func NewKafka(brokers []string, c *sarama.Config, h ConsumerGroupHandler, provider string) (k *Kafka, err error)

func (*Kafka) Append

func (e *Kafka) Append(opts ...storage.Option) error

func (*Kafka) Register

func (e *Kafka) Register(opts ...storage.Option)

func (*Kafka) Run

func (e *Kafka) Run(ctx context.Context)

func (*Kafka) Shutdown

func (e *Kafka) Shutdown()

func (*Kafka) String

func (*Kafka) String() string

type KafkaRunReader

type KafkaRunReader struct {
	Topic   string
	GroupID string
	Func    storage.ConsumerFunc
}

type Memory

type Memory struct {
	PoolNum uint
	// contains filtered or unexported fields
}

func NewMemory

func NewMemory(poolNum uint) *Memory

NewMemory 内存模式

func (*Memory) Append

func (m *Memory) Append(opts ...storage.Option) error

func (*Memory) Register

func (m *Memory) Register(opts ...storage.Option)

func (*Memory) Run

func (m *Memory) Run(context.Context)

func (*Memory) Shutdown

func (m *Memory) Shutdown()

func (*Memory) String

func (*Memory) String() string

type Message

type Message struct {
	redisqueue.Message
	ErrorCount int
	// contains filtered or unexported fields
}

func (*Message) GetContext

func (m *Message) GetContext() context.Context

func (*Message) GetErrorCount

func (m *Message) GetErrorCount() int

func (*Message) GetID

func (m *Message) GetID() string

func (*Message) GetPrefix

func (m *Message) GetPrefix() (prefix string)

func (*Message) GetStream

func (m *Message) GetStream() string

func (*Message) GetValues

func (m *Message) GetValues() map[string]interface{}

func (*Message) SetContext

func (m *Message) SetContext(ctx context.Context)

func (*Message) SetErrorCount

func (m *Message) SetErrorCount(count int)

func (*Message) SetID

func (m *Message) SetID(id string)

func (*Message) SetPrefix

func (m *Message) SetPrefix(prefix string)

func (*Message) SetStream

func (m *Message) SetStream(stream string)

func (*Message) SetValues

func (m *Message) SetValues(values map[string]interface{})

type MessageHandler

type MessageHandler struct {
	// contains filtered or unexported fields
}

func (*MessageHandler) Cleanup

func (h *MessageHandler) Cleanup(sarama.ConsumerGroupSession) error

func (*MessageHandler) ConsumeClaim

func (h *MessageHandler) ConsumeClaim(s sarama.ConsumerGroupSession, c sarama.ConsumerGroupClaim) error

func (*MessageHandler) SetConsumerFunc

func (h *MessageHandler) SetConsumerFunc(f storage.ConsumerFunc)

func (*MessageHandler) Setup

func (h *MessageHandler) Setup(s sarama.ConsumerGroupSession) error

type NSQ

type NSQ struct {
	// contains filtered or unexported fields
}

func NewNSQ

func NewNSQ(cfg *nsq.Config, lookup, adminAddr string, addresses ...string) (*NSQ, error)

NewNSQ nsq模式 只能监听一个channel

func (*NSQ) Append

func (e *NSQ) Append(opts ...storage.Option) error

Append 消息入生产者

func (*NSQ) Register

func (e *NSQ) Register(opts ...storage.Option)

Register 监听消费者

func (*NSQ) Run

func (e *NSQ) Run(context.Context)

func (*NSQ) Shutdown

func (e *NSQ) Shutdown()

func (*NSQ) String

func (*NSQ) String() string

String 字符串类型

type NodesResp

type NodesResp struct {
	Nodes   []*peerInfo `json:"nodes"`
	Message string      `json:"message"`
}

type Redis

type Redis struct {
	// contains filtered or unexported fields
}

Redis cache implement

func NewRedis

func NewRedis(
	producerOptions *redisqueue.ProducerOptions,
	consumerOptions *redisqueue.ConsumerOptions,
) (*Redis, error)

NewRedis redis模式

func (*Redis) Append

func (r *Redis) Append(opts ...storage.Option) error

func (*Redis) Register

func (r *Redis) Register(opts ...storage.Option)

func (*Redis) Run

func (r *Redis) Run(context.Context)

func (*Redis) Shutdown

func (r *Redis) Shutdown()

func (*Redis) String

func (*Redis) String() string

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)