kafka

KO | EN

gitlite — search

#python #hacktoberfest #ai #typescript #llm #machine-learning #deep-learning #linux #macos #cli #awesome-list #vue #security #openai

(★ 212)

Go driver for Kafka

File Explorer

Download Latest Version (.zip)

- config.yml
- - pr.yml
- ca.cnf
- ca.crt
- generate_certificates.sh
- oats.cnf
- oats.crt
- oats.extensions.cnf
- oats.key
- - broker_test.go
  - cluster.go
  - cluster_test.go
  - README.md
- - broker.go
  - broker_test.go
  - doc.go
  - example_test.go
  - server.go
- - doc.go
  - errors.go
  - messages.go
  - messages_go_1.7_test.go
  - messages_go_1.8_test.go
  - messages_test.go
  - serialization.go
  - serialization_test.go
  - snappy.go
  - snappy_test.go
  - utils.go
- broker.go
- broker_test.go
- connection.go
- connection_test.go
- distributing_producer.go
- distributing_producer_test.go
- doc.go
- example_test.go
- go.mod
- go.sum
- log.go
- multiplexer.go
- multiplexer_test.go
- server_test.go
.gitignore
CODEOWNERS
go.mod
LICENSE
README.md

// repository documentation

Was this content helpful?

★ 0 (0 ratings)

[![Tests status](https://github.com/optiopay/kafka/workflows/Kafka/badge.svg)](https://github.com/optiopay/kafka/actions?query=workflow%3AKafka)
[![GoDoc](https://godoc.org/github.com/optiopay/kafka/v2?status.png)](https://godoc.org/github.com/optiopay/kafka/v2)

# Kafka

Kafka is Go client library for [Apache Kafka](https://kafka.apache.org/)
server, released under [MIT license](LICENSE]).

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

* [godoc](https://godoc.org/github.com/optiopay/kafka/v2) generated documentation,
* [code examples](https://godoc.org/github.com/optiopay/kafka/v2#pkg-examples)

## Example

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

```go
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)
}
```

kafka

File Explorer

Today's picks

Was this content helpful?

Recent Feedback

kafka

File Explorer

Today's picks

Was this content helpful?

Recent Feedback

Download README