client

package

Go to main page

Versions in this module

v1

v1.7.0

Dec 12, 2018

v1.6.0

Oct 4, 2018

v1.5.0

Aug 24, 2018

v1.4.1

Jul 23, 2018

v1.4.0

Jul 16, 2018

v1.3.0

May 24, 2018

v1.2.0

Mar 22, 2018

v1.1.0

Feb 7, 2018

v1.0.8

Jan 22, 2018

Changes in this version

type Config

+ func (ref *Config) SetTLS(tlsConfig *tls.Config) (err error)

v1.0.7

Nov 21, 2017

v1.0.6

Oct 30, 2017

Changes in this version

type ConsumerMessage

+ PrevValue []byte

+ func (cm *ConsumerMessage) GetPrevValue() []byte

type ProducerMessage

+ func (pm *ProducerMessage) GetPrevValue() []byte

type ProtoConsumerMessage

+ func (cm *ProtoConsumerMessage) GetPrevValue(msg proto.Message) (prevValueExist bool, err error)

type ProtoProducerMessage

+ func (ppm *ProtoProducerMessage) GetPrevValue(msg proto.Message) (prevValueExist bool, err error)

v1.0.5

Oct 17, 2017

Changes in this version

type Consumer

+ SConsumer sarama.Consumer

+ func (ref *Consumer) StartConsumerHandlers()

+ func (ref *Consumer) StartConsumerManualHandlers(partitionConsumer sarama.PartitionConsumer)

v1.0.4

Sep 26, 2017

Changes in this version

+ func NewClient(config *Config, partitioner string) (sarama.Client, error)

type AsyncProducer

+ func (ref *AsyncProducer) SendMsgToPartition(topic string, partition int32, key Encoder, msg Encoder, metadata interface{})

type Consumer

+ func (ref *Consumer) ConsumerErrorHandler(in <-chan *sarama.ConsumerError)

+ func (ref *Consumer) MessageHandler(in <-chan *sarama.ConsumerMessage)

type SyncProducer

+ func (ref *SyncProducer) SendMsgToPartition(topic string, partition int32, key sarama.Encoder, msg sarama.Encoder) (*ProducerMessage, error)

v1.0.3

Sep 8, 2017

Changes in this version

+ const Hash

+ const Manual

+ const Random

type ConsumerMessage

+ func (cm *ConsumerMessage) GetOffset() int64

+ func (cm *ConsumerMessage) GetPartition() int32

type ProducerMessage

+ func (pm *ProducerMessage) GetOffset() int64

+ func (pm *ProducerMessage) GetPartition() int32

type ProtoConsumerMessage

+ func (cm *ProtoConsumerMessage) GetOffset() int64

+ func (cm *ProtoConsumerMessage) GetPartition() int32

type ProtoProducerMessage

+ func (ppm *ProtoProducerMessage) GetOffset() int64

+ func (ppm *ProtoProducerMessage) GetPartition() int32

v1.0.2

Aug 28, 2017

Changes in this version

+ type AsyncProducer struct

+ Client sarama.Client

+ Config *Config

+ Partition int32

+ Producer sarama.AsyncProducer

+ func GetAsyncProducerMock(t mocks.ErrorReporter) (*AsyncProducer, *mocks.AsyncProducer)

+ func NewAsyncProducer(config *Config, wg *sync.WaitGroup) (*AsyncProducer, error)

+ func (ref *AsyncProducer) Close(async ...bool) error

+ func (ref *AsyncProducer) GetCloseChannel() <-chan struct{}

+ func (ref *AsyncProducer) IsClosed() bool

+ func (ref *AsyncProducer) SendMsg(topic string, key Encoder, msg Encoder, metadata interface{})

+ func (ref *AsyncProducer) SendMsgByte(topic string, key []byte, msg []byte, metadata interface{})

+ func (ref *AsyncProducer) WaitForClose()

+ type Config struct

+ Brokers []string

+ Cancel context.CancelFunc

+ Context context.Context

+ Debug bool

+ ErrorChan chan *ProducerError

+ GroupID string

+ InitialOffset int64

+ Partition int32

+ Partitioner sarama.PartitionerConstructor

+ RecvError bool

+ RecvErrorChan chan error

+ RecvMessageChan chan *ConsumerMessage

+ RecvNotification bool

+ RecvNotificationChan chan *cluster.Notification

+ RequiredAcks RequiredAcks

+ SendError bool

+ SendSuccess bool

+ SuccessChan chan *ProducerMessage

+ Topics []string

+ func NewConfig(log logging.Logger) *Config

+ func (ref *Config) ConsumerConfig() *cluster.Config

+ func (ref *Config) ProducerConfig() *sarama.Config

+ func (ref *Config) SetAcks(acks RequiredAcks)

+ func (ref *Config) SetBrokers(brokers ...string)

+ func (ref *Config) SetDebug(val bool)

+ func (ref *Config) SetErrorChan(val chan *ProducerError)

+ func (ref *Config) SetGroup(id string)

+ func (ref *Config) SetInitialOffset(offset int64)

+ func (ref *Config) SetPartition(val int32)

+ func (ref *Config) SetPartitioner(val string)

+ func (ref *Config) SetRecvError(val bool)

+ func (ref *Config) SetRecvErrorChan(val chan error)

+ func (ref *Config) SetRecvMessageChan(val chan *ConsumerMessage)

+ func (ref *Config) SetRecvNotification(val bool)

+ func (ref *Config) SetRecvNotificationChan(val chan *cluster.Notification)

+ func (ref *Config) SetSendError(val bool)

+ func (ref *Config) SetSendSuccess(val bool)

+ func (ref *Config) SetSuccessChan(val chan *ProducerMessage)

+ func (ref *Config) SetTopics(topics string)

+ func (ref *Config) ValidateAsyncProducerConfig() error

+ func (ref *Config) ValidateConsumerConfig() error

+ func (ref *Config) ValidateSyncProducerConfig() error

+ type Consumer struct

+ Client sarama.Client

+ Config *Config

+ Consumer clusterConsumer

+ func GetConsumerMock(t mocks.ErrorReporter) *Consumer

+ func NewConsumer(config *Config, wg *sync.WaitGroup) (*Consumer, error)

+ func (ref *Consumer) Close() error

+ func (ref *Consumer) CommitOffsets() error

+ func (ref *Consumer) GetCloseChannel() <-chan struct{}

+ func (ref *Consumer) IsClosed() bool

+ func (ref *Consumer) MarkOffset(msg *ConsumerMessage, metadata string)

+ func (ref *Consumer) MarkPartitionOffset(topic string, partition int32, offset int64, metadata string)

+ func (ref *Consumer) PrintNotification(note map[string][]int32)

+ func (ref *Consumer) Subscriptions() map[string][]int32

+ func (ref *Consumer) WaitForClose()

+ type ConsumerMessage struct

+ Key []byte

+ Offset int64

+ Partition int32

+ Timestamp time.Time

+ Topic string

+ Value []byte

+ func (cm *ConsumerMessage) GetKey() string

+ func (cm *ConsumerMessage) GetTopic() string

+ func (cm *ConsumerMessage) GetValue() []byte

+ type Encoder interface

+ type ProducerError struct

+ Err error

+ func (ref *ProducerError) Error() error

+ func (ref *ProducerError) String() string

+ type ProducerMessage struct

+ Key Encoder

+ Metadata interface{}

+ Offset int64

+ Partition int32

+ Topic string

+ Value Encoder

+ func (pm *ProducerMessage) GetKey() string

+ func (pm *ProducerMessage) GetTopic() string

+ func (pm *ProducerMessage) GetValue() []byte

+ func (pm *ProducerMessage) String() string

+ type ProtoConsumerMessage struct

+ func NewProtoConsumerMessage(msg *ConsumerMessage, serializer keyval.Serializer) *ProtoConsumerMessage

+ func (cm *ProtoConsumerMessage) GetKey() string

+ func (cm *ProtoConsumerMessage) GetTopic() string

+ func (cm *ProtoConsumerMessage) GetValue(msg proto.Message) error

+ type ProtoProducerMessage struct

+ Serializer keyval.Serializer

+ func (ppm *ProtoProducerMessage) GetKey() string

+ func (ppm *ProtoProducerMessage) GetTopic() string

+ func (ppm *ProtoProducerMessage) GetValue(msg proto.Message) error

+ type ProtoProducerMessageErr struct

+ Err error

+ func (pme *ProtoProducerMessageErr) Error() error

+ type RequiredAcks int16

+ const AcksUnset

+ const NoResponse

+ const WaitForAll

+ const WaitForLocal

+ type SyncProducer struct

+ Client sarama.Client

+ Config *Config

+ Partition int32

+ Producer sarama.SyncProducer

+ func GetSyncProducerMock(t mocks.ErrorReporter) (*SyncProducer, *mocks.SyncProducer)

+ func NewSyncProducer(config *Config, wg *sync.WaitGroup) (*SyncProducer, error)

+ func (ref *SyncProducer) Close() error

+ func (ref *SyncProducer) GetCloseChannel() <-chan struct{}

+ func (ref *SyncProducer) IsClosed() bool

+ func (ref *SyncProducer) SendMsg(topic string, key sarama.Encoder, msg sarama.Encoder) (*ProducerMessage, error)

+ func (ref *SyncProducer) SendMsgByte(topic string, key []byte, msg []byte) (*ProducerMessage, error)

+ func (ref *SyncProducer) WaitForClose()

Incompatible versions in this module

v2.2.0+incompatible

Aug 23, 2019

v2.1.0+incompatible

May 9, 2019

v2.0.0+incompatible

Apr 2, 2019

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