kafka

Documentation

Index

• func Initialize(host string, port int, prefix string, hasProducer bool, ...)
• func Recover()
• func SendMessageToTopic(message, topic string)
• type Client
  • func NewClient() (client *Client)
  • func (client *Client) Initialize(host string, port int, prefix string, hasProducer bool, ...)
  • func (client *Client) SendMessageToTopic(message, topic string)
• type MessageConsumer

Functions

func Initialize

func Initialize(host string, port int, prefix string, hasProducer bool, consumers []MessageConsumer)

Initialize : initializes the kafka producer/consumer.

func Recover

func Recover()

Recover : recovers from a panic, use this function to the consumers.

func SendMessageToTopic

func SendMessageToTopic(message, topic string)

SendMessageToTopic : sends a message to a topic.

Types

type Client

type Client struct {
	HasProducer     bool
	Producer        sarama.AsyncProducer
	ProducerSignals chan os.Signal
	TopicsPrefix    string
}

Client : a kafka client. Could be a message producer.

var Main *Client

Main : main client of the application.

func NewClient

func NewClient() (client *Client)

NewClient : initialize a new Client struct.

func (*Client) Initialize

func (client *Client) Initialize(host string, port int, prefix string, hasProducer bool, consumers []MessageConsumer)

Initialize initializes the kafka producer/consumer.

func (*Client) SendMessageToTopic

func (client *Client) SendMessageToTopic(message, topic string)

SendMessageToTopic sends a message to a topic.

type MessageConsumer

type MessageConsumer struct {
	Topic  string
	Action func(value []byte)
}

MessageConsumer : a message consumer consumes messages from a queue 'topic' and executes 'action'