kafka

Documentation

Index

• func SetConsumeReturnDuration(dur ConsumerWaitDuration) func(*Kafka) error
• func SetNumberMaximumLogMessages(mc MaxLogMessages) func(*Kafka) error
• func SetNumberOfPartitions(pc PartitionCount) func(*Kafka) error
• func SetNumberOfReplicas(rc ReplicaCount) func(*Kafka) error
• type BrokerAddresses
• type ConsumerWaitDuration
• type Kafka
  • func New(ver Version, brk BrokerAddresses, options ...func(*Kafka) error) (*Kafka, error)
  • func (k *Kafka) AppendLog(txID string, data string) error
  • func (k *Kafka) GetTxLogs(txID string) ([]string, error)
  • func (k *Kafka) SetDefaults()
  • func (k *Kafka) TxIDAlreadyExists(s string) (bool, error)
• type MaxLogMessages
• type PartitionCount
• type ReplicaCount
• type Topic
• type Version

Functions

func SetConsumeReturnDuration

func SetConsumeReturnDuration(dur ConsumerWaitDuration) func(*Kafka) error

SetConsumeReturnDuration is the functional option to set ConsumerWaitDuration

func SetNumberMaximumLogMessages

func SetNumberMaximumLogMessages(mc MaxLogMessages) func(*Kafka) error

SetNumberMaximumLogMessages is the functional option to set MaxLogMessages

func SetNumberOfPartitions

func SetNumberOfPartitions(pc PartitionCount) func(*Kafka) error

SetNumberOfPartitions is the functional option to set PartitionCount

func SetNumberOfReplicas

func SetNumberOfReplicas(rc ReplicaCount) func(*Kafka) error

SetNumberOfReplicas is the functional option to set ReplicaCount

Types

type BrokerAddresses

type BrokerAddresses []string

BrokerAddresses is the list of broker addresses e.g. ["192.168.1.19:9092", "192.168.1.21:9092"]

type ConsumerWaitDuration

type ConsumerWaitDuration time.Duration

ConsumerWaitDuration is the timeout duration for which consumer will wait for the message

type Kafka

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

Kafka object provides storage functions to provide persistence medium for Saga. Saga is as reliable as it's persistence medium.

func New

func New(ver Version, brk BrokerAddresses, options ...func(*Kafka) error) (*Kafka, error)

New to create a new Kafka object. It accepts functional options to support maximum customization.

func (*Kafka) AppendLog

func (k *Kafka) AppendLog(txID string, data string) error

AppendLog

func (*Kafka) GetTxLogs

func (k *Kafka) GetTxLogs(txID string) ([]string, error)

GetTxLogs to get Tx logs

func (*Kafka) SetDefaults

func (k *Kafka) SetDefaults()

SetDefaults sets sensible defaults to the Kafka config

func (*Kafka) TxIDAlreadyExists

func (k *Kafka) TxIDAlreadyExists(s string) (bool, error)

type MaxLogMessages

type MaxLogMessages int

MaxLogMessages is the maximum number of logs messages that will be published to Kafka in a single Transaction.

type PartitionCount

type PartitionCount int32

PartitionCount is the number of partitions for any topic that'll be created by saga

type ReplicaCount

type ReplicaCount int16

ReplicaCount is the number of replicas for any topic that'll be created by saga

type Topic

type Topic interface {
	IsTopicAlreadyCreated(topicName string) (exists bool, err error)
	GetAllTopics() (topicNames []string, err error)
	CreateTopic(topicName string, partitionCount int32, replicaCount int16) (err error)
	DeleteTopic(topicName string) (err error)
}

Topic provides kafka-topic management functions.

type Version

type Version string

Version is the Kafka version e.g. "2.7.0"