rkafka

README

rkafka

Quick Start!

package main
import (
	"bytes"
	"context"
	"fmt"
	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/dawei101/gor/rkafka"
	"strconv"
	"strings"
	"time"
)

func main() {
	manager := rkafka.New()

	manager.OnProcess("userLogin", func(msg *kafka.Message, ctx context.Context) error {
		println(string(msg.Value))
		return nil
	})
	//
	manager.Run()
}

Documentation

Index

• type Consumer
  • func (c *Consumer) Run()
• type ConsumerConf
• type Limiter
  • func (l *Limiter) Run(f func())
• type Manager
  • func New(managerConf *ManagerConf) *Manager
  • func (m *Manager) OnProcesses(name string, fn OnProcess)
  • func (m *Manager) Run()
  • func (m *Manager) Stop()
• type ManagerConf
• type OnError
• type OnProcess

Types

type Consumer

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

func (*Consumer) Run

func (c *Consumer) Run()

type ConsumerConf

type ConsumerConf struct {
	ClientID string `yaml:"clientID"`
	GroupID  string `yaml:"groupID"`
	Server   string `yaml:"server"`
	Topic    string `yaml:"topic"`
	Limit    int    `yaml:"limit"`
}

type Limiter

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

func (*Limiter) Run

func (l *Limiter) Run(f func())

type Manager

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

func New

func New(managerConf *ManagerConf) *Manager

func (*Manager) OnProcesses

func (m *Manager) OnProcesses(name string, fn OnProcess)

func (*Manager) Run

func (m *Manager) Run()

func (*Manager) Stop

func (m *Manager) Stop()

type ManagerConf

type ManagerConf struct {
	Consumers map[string]ConsumerConf
}

type OnError

type OnError func(err error, ctx context.Context)

type OnProcess

type OnProcess func(msg *kafka.Message, ctx context.Context)