subscriber

Documentation

Overview

Package subscriber implements Kafka Consumer for Go.

Example (Minimal)

package main

import (
	"fmt"
	"log"

	"github.com/dimmyjr/goku/message"
	"github.com/dimmyjr/goku/subscriber"
	"github.com/dimmyjr/goku/types"
)

func main() {
	kafkaURL := "localhost:9092"
	topic := "topic-segmentio"
	groupID := "tests"

	consumer, err := subscriber.NewConsumer([]string{kafkaURL}, topic, groupID, types.Segmentio)

	if err != nil {
		log.Fatal("error to initialize producer")
	}

	consumer.Subscribe(read)
}

func read(msg *message.Message) error {
	fmt.Print(&msg.Value)

	return nil
}

Index

• Variables
• func NewConsumer(kafkaURLs []string, topic, groupID string, provider types.Provider) (consumer.Consumer, error)
• type Subscriber
  • func (s Subscriber) Close()
  • func (s Subscriber) Subscribe(f func(message *message.Message) error)

Examples

• Package (Minimal)

Variables

var ErrUndefinedProvider = errors.New("undefinedProvider")

ErrUndefinedProvider : undefined provider error.

Functions

func NewConsumer

func NewConsumer(kafkaURLs []string, topic, groupID string, provider types.Provider) (consumer.Consumer, error)

NewConsumer create a Kafka Consumer with the chosen provider return: a new consumer or error.

Types

type Subscriber

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

func (Subscriber) Close

func (s Subscriber) Close()

Close finish consumer and stop read messages.

func (Subscriber) Subscribe

func (s Subscriber) Subscribe(f func(message *message.Message) error)

Subscribe reads and return the message from the Topic.