kafka

package pubsub/kafka

package pubsub/kafka provides a set of interfaces for publishing and subscribing events from Kafka.

Usage

Basic

import (
	"github.com/Shopify/sarama"

	"github.com/mo-work/mo-kit/pubsub/kafka"
)

func main() {
  consumer, err := kafka.NewConsumer(kafkaBrokers, kafkaGroup, "2.2.0", log.With(logger, "component", "kafka"), nr)
	if err != nil {
		panic(fmt.Sprintf("Error initializing kafka: %s", err))
	}
	handlers := no.NewEventHandlers(us)

	errs := make(chan error)

	go func() {
		consumer.Start(kafkaTopics, handlers)
	}()
}

Create event handlers and assign to the module

With topic and key given

import (
	"github.com/Shopify/sarama"

	"github.com/mo-work/mo-kit/pubsub/kafka"
)


func NewEventHandlers(s Service) *kafka.EventHandlers {
	handlers := make(kafka.EventHandlers)
	handlers["moment_created"] = momentHandler(s)

	return &handlers
}

func momentHandler(s Service) kafka.EventHandler {
	return func(message *sarama.ConsumerMessage) error {
    fmt.Println("Message has been received")
		return nil
	}
}

With only topic given

import (
	"github.com/Shopify/sarama"

	"github.com/mo-work/mo-kit/pubsub/kafka"
)


func NewEventHandlers(s Service) *kafka.EventHandlers {
	handlers := make(kafka.EventHandlers)
	handlers["moment"] = momentHandler(s)

	return &handlers
}

func momentHandler(s Service) kafka.EventHandler {
	return func(message *sarama.ConsumerMessage) error {
    fmt.Println("Message has been received")
		return nil
	}
}

With only topic and topic / key combination

import (
	"github.com/Shopify/sarama"

	"github.com/mo-work/mo-kit/pubsub/kafka"
)


func NewEventHandlers(s Service) *kafka.EventHandlers {
	handlers := make(kafka.EventHandlers)
  handlers["moment_unique"] = momentUniqueHandler(s)
	handlers["moment"] = momentHandler(s)


	return &handlers
}

func momentHandler(s Service) kafka.EventHandler {
	return func(message *sarama.ConsumerMessage) error {
    fmt.Println("Message has been received")
		return nil
	}
}

func momentUniqueHandler(s Service) kafka.EventHandler {
	return func(message *sarama.ConsumerMessage) error {
    fmt.Println("Moment Unique Event has been received")
		return nil
	}
}