consumer

package

v1.0.0 Latest Latest Go to latest Published: Mar 28, 2023 License: MIT Imports: 26 Imported by: 0

Details

Valid go.mod file

The Go module system was introduced in Go 1.11 and is the official dependency management solution for Go.
Redistributable license

Redistributable licenses place minimal restrictions on how software can be used, modified, and redistributed.
Tagged version

Modules with tagged versions give importers more predictable builds.
Stable version

When a project reaches major version v1 it is considered stable.
Learn more about best practices

Repository

github.com/bitleak/kaproxy

Links

Open Source Insights

Documentation ¶

Index ¶

Constants
Variables
type Consumer
- func NewConsumer(saramaClient sarama.Client, conf *config.Config) (*Consumer, error)
type GroupMetadata
type Puller
type TopicUnackManager

Constants ¶

View Source

const (
	SemanticAtMostOnce  = "atMostOnce"
	SemanticAtLeastOnce = "atLeastOnce"

	OwnerName = "kaproxy"
)

View Source

const (
	PeerProxyID = "__peer_proxy__"
)

Variables ¶

View Source

var (
	ErrGroupStopped         = errors.New("consumer group was stopped")
	ErrGroupNotFound        = errors.New("consumer group not found")
	ErrTopicNotFound        = errors.New("topic not found in this consumer group")
	ErrNoMessage            = errors.New("no message in brokers")
	ErrUnackManagerStopped  = errors.New("unack manager was stopped")
	ErrUnackMessageNotFound = errors.New("unack message not found")
	ErrGroupNotAllowACK     = errors.New("consumer group is not allowed to be acknowledged")
	ErrNoPartition          = errors.New("consumer group doesn't claim any partition")
	ErrNoPeer               = errors.New("no peer can pull message")
)

Functions ¶

This section is empty.

Types ¶

type Consumer ¶

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

func NewConsumer ¶

func NewConsumer(saramaClient sarama.Client, conf *config.Config) (*Consumer, error)

func (*Consumer) ACK ¶

func (c *Consumer) ACK(token, group, topic string, partition int32, offset int64) error

func (*Consumer) Consume ¶

func (c *Consumer) Consume(group, topic string, timeout, ttr time.Duration) (message *sarama.ConsumerMessage, err error)

func (*Consumer) Export ¶

func (c *Consumer) Export(group string) (map[string]interface{}, error)

func (*Consumer) GetConsumerGroup ¶

func (c *Consumer) GetConsumerGroup(group string) *consumerGroup

func (*Consumer) GetConsumerGroupState ¶

func (c *Consumer) GetConsumerGroupState(group string) (string, error)

func (*Consumer) GetMetadataByName ¶

func (c *Consumer) GetMetadataByName(group string) *GroupMetadata

func (*Consumer) ListConsumerGroup ¶

func (c *Consumer) ListConsumerGroup() []string

func (*Consumer) ReleaseConsumerGroups ¶

func (c *Consumer) ReleaseConsumerGroups()

func (*Consumer) Start ¶

func (c *Consumer) Start() error

func (*Consumer) Stop ¶

func (c *Consumer) Stop()

func (*Consumer) UpdateConsumerGroupState ¶

func (c *Consumer) UpdateConsumerGroupState(group string, stopped bool) error

type GroupMetadata ¶

type GroupMetadata struct {
	Owner     string                `json:"owner"`
	Topics    []string              `json:"topics"`
	Semantics string                `json:"semantics"`
	Consumer  config.ConsumerConfig `json:"consumer"`
	Stopped   bool                  `json:"stopped"`
	// contains filtered or unexported fields
}

type Puller ¶

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

func (*Puller) Close ¶

func (p *Puller) Close() error

func (*Puller) PullFromPeers ¶

func (p *Puller) PullFromPeers(group, topic, token string, timeout, ttr uint64, noPartition bool) (*sarama.ConsumerMessage, error)

func (*Puller) Start ¶

func (p *Puller) Start() error

type TopicUnackManager ¶

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

Source Files ¶

View all Source files

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL