kafka

Documentation

Index

• func InstallMiddleware(handle ...ProducerHandler)
• type Consumer
  • func GetConsumer() Consumer
• type ConsumerImpl
  • func (c *ConsumerImpl) Booting(bootManager freedom.BootManager)
  • func (c *ConsumerImpl) Close() error
  • func (c *ConsumerImpl) Restart() error
  • func (c *ConsumerImpl) SetProxyTimeout(timeout time.Duration)
  • func (c *ConsumerImpl) SetRateLimit(rate int)
  • func (c *ConsumerImpl) Start(addrs []string, groupID string, config *sarama.Config, proxyAddr string, ...)
• type Msg
  • func (msg *Msg) GetExecution() error
  • func (msg *Msg) GetHeader() map[string]interface{}
  • func (msg *Msg) GetMessageKey() string
  • func (msg *Msg) IsStopped() bool
  • func (msg *Msg) Next()
  • func (msg *Msg) Publish() error
  • func (msg *Msg) SetHeader(head map[string]interface{}) *Msg
  • func (msg *Msg) SetMessageKey(key string) *Msg
  • func (msg *Msg) Stop() *Msg
• type Producer
  • func GetProducer() Producer
• type ProducerHandler
• type ProducerImpl
  • func (pi *ProducerImpl) Booting(bootManager freedom.BootManager)
  • func (pi *ProducerImpl) Close() error
  • func (pi *ProducerImpl) NewMsg(topic string, content []byte) *Msg
  • func (pi *ProducerImpl) Restart() error
  • func (pi *ProducerImpl) Start(addrs []string, config *sarama.Config)

Functions

func InstallMiddleware

func InstallMiddleware(handle ...ProducerHandler)

InstallMiddleware Install the middleware.. You can control the publishing of messages by installing middleware.

Types

type Consumer

type Consumer interface {
	// Start pass in the relevant address, configuration.
	Start(addrs []string, groupID string, config *sarama.Config, proxyAddr string, proxyH2C bool)
	// Restart the connection.
	Restart() error
	// Close the connection.
	Close() error
	// Set the limit flow per second.
	SetRateLimit(rate int)
	// Set the HTTP agent to time out.
	SetProxyTimeout(time.Duration)
}

Consumer Kafka Consumer interface definition.

func GetConsumer

func GetConsumer() Consumer

GetConsumer Returns the Consumer instance.

type ConsumerImpl

type ConsumerImpl struct {
	freedom.Infra
	// contains filtered or unexported fields
}

ConsumerImpl Kafka Consumer implementation.

func (*ConsumerImpl) Booting

func (c *ConsumerImpl) Booting(bootManager freedom.BootManager)

Booting The method of overriding the component . The single-case component initiates a callback.

func (*ConsumerImpl) Close

func (c *ConsumerImpl) Close() error

Close the connection.

func (*ConsumerImpl) Restart

func (c *ConsumerImpl) Restart() error

Restart the connection.

func (*ConsumerImpl) SetProxyTimeout

func (c *ConsumerImpl) SetProxyTimeout(timeout time.Duration)

SetProxyTimeout Set the HTTP agent to time out.

func (*ConsumerImpl) SetRateLimit

func (c *ConsumerImpl) SetRateLimit(rate int)

SetRateLimit Set the limit flow per second.

func (*ConsumerImpl) Start

func (c *ConsumerImpl) Start(addrs []string, groupID string, config *sarama.Config, proxyAddr string, proxyH2C bool)

Start pass in the relevant address, configuration.

type Msg

type Msg struct {
	Topic string

	Content []byte
	// contains filtered or unexported fields
}

Msg Kafka Message.

func (*Msg) GetExecution

func (msg *Msg) GetExecution() error

GetExecution Get the results of the execution .

func (*Msg) GetHeader

func (msg *Msg) GetHeader() map[string]interface{}

GetHeader .

func (*Msg) GetMessageKey

func (msg *Msg) GetMessageKey() string

GetMessageKey Get kafka key.

func (*Msg) IsStopped

func (msg *Msg) IsStopped() bool

IsStopped whether it has stopped.

func (*Msg) Next

func (msg *Msg) Next()

Next Perform the next step, typically for the control of middleware.

func (*Msg) Publish

func (msg *Msg) Publish() error

Publish this message.

func (*Msg) SetHeader

func (msg *Msg) SetHeader(head map[string]interface{}) *Msg

SetHeader set up HTTP Header.

func (*Msg) SetMessageKey

func (msg *Msg) SetMessageKey(key string) *Msg

SetMessageKey Set kafka key.

func (*Msg) Stop

func (msg *Msg) Stop() *Msg

Stop execution, typically used for middleware control

type Producer

type Producer interface {
	// Create a new message
	NewMsg(topic string, content []byte) *Msg
	// Start pass in the relevant address, configuration.
	Start(addrs []string, config *sarama.Config)
	// Restart the connection.
	Restart() error
}

Producer The producer's interface definition.

func GetProducer

func GetProducer() Producer

GetProducer Gets an instance of the producer.

type ProducerHandler

type ProducerHandler func(*Msg)

ProducerHandler The function declaration of the Kafka Producer middleware..

type ProducerImpl

type ProducerImpl struct {
	freedom.Infra
	// contains filtered or unexported fields
}

ProducerImpl The realization of the producer.

func (*ProducerImpl) Booting

func (pi *ProducerImpl) Booting(bootManager freedom.BootManager)

Booting The method of overriding the component . The single-case component initiates a callback.

func (*ProducerImpl) Close

func (pi *ProducerImpl) Close() error

Close .

func (*ProducerImpl) NewMsg

func (pi *ProducerImpl) NewMsg(topic string, content []byte) *Msg

NewMsg Create a new message.

func (*ProducerImpl) Restart

func (pi *ProducerImpl) Restart() error

Restart the connection.

func (*ProducerImpl) Start

func (pi *ProducerImpl) Start(addrs []string, config *sarama.Config)

Start pass in the relevant address, configuration.