kafka

by optiopay

optiopay / kafka

Go driver for Kafka

212 Stars 53 Forks Last release: 12 months ago (v2.1.1) MIT License 429 Commits 11 Releases

Available items

No Items, yet!

The developer of this repository has not created any items for sale yet. Need a bug fixed? Help with integration? A different license? Create a request here:

Tests status GoDoc

Kafka

Kafka is Go client library for Apache Kafka server, released under MIT license.

Kafka provides minimal abstraction over wire protocol, support for transparent failover and easy to use blocking API.

Example

Write all messages from stdin to kafka and print all messages from kafka topic to stdout.

package main

import ( "bufio" "log" "os" "strings"

"github.com/optiopay/kafka/v2"
"github.com/optiopay/kafka/v2/proto"

)

const ( topic = "my-messages" partition = 0 )

var kafkaAddrs = []string{"localhost:9092", "localhost:9093"}

// printConsumed read messages from kafka and print them out func printConsumed(broker kafka.Client) { conf := kafka.NewConsumerConf(topic, partition) conf.StartOffset = kafka.StartOffsetNewest consumer, err := broker.Consumer(conf) if err != nil { log.Fatalf("cannot create kafka consumer for %s:%d: %s", topic, partition, err) }

for {
    msg, err := consumer.Consume()
    if err != nil {
        if err != kafka.ErrNoData {
            log.Printf("cannot consume %q topic message: %s", topic, err)
        }
        break
    }
    log.Printf("message %d: %s", msg.Offset, msg.Value)
}
log.Print("consumer quit")

}

// produceStdin read stdin and send every non empty line as message func produceStdin(broker kafka.Client) { producer := broker.Producer(kafka.NewProducerConf()) input := bufio.NewReader(os.Stdin) for { line, err := input.ReadString('\n') if err != nil { log.Fatalf("input error: %s", err) } line = strings.TrimSpace(line) if line == "" { continue }

    msg := &proto.Message{Value: []byte(line)}
    if _, err := producer.Produce(topic, partition, msg); err != nil {
        log.Fatalf("cannot produce message to %s:%d: %s", topic, partition, err)
    }
}

}

func main() { conf := kafka.NewBrokerConf("test-client") conf.AllowTopicCreation = true

// connect to kafka cluster
broker, err := kafka.Dial(kafkaAddrs, conf)
if err != nil {
    log.Fatalf("cannot connect to kafka cluster: %s", err)
}
defer broker.Close()

go printConsumed(broker)
produceStdin(broker)

}

We use cookies. If you continue to browse the site, you agree to the use of cookies. For more information on our use of cookies please see our Privacy Policy.