properties

Documentation

Index

• type Admin
  • func (p Admin) GetClientId() string
  • func (p Admin) GetSecurityProtocol() string
  • func (p Admin) GetTls() *Tls
• type Client
  • func NewClient(loader config.Loader) (*Client, error)
  • func (p *Client) PostBinding() error
  • func (p Client) Prefix() string
• type Consumer
  • func (p Consumer) GetClientId() string
  • func (p Consumer) GetSecurityProtocol() string
  • func (p Consumer) GetTls() *Tls
• type EventProducer
  • func NewEventProducer(loader config.Loader) (*EventProducer, error)
  • func (p EventProducer) Prefix() string
• type EventTopic
• type KafkaConsumer
  • func NewKafkaConsumer(loader config.Loader) (*KafkaConsumer, error)
  • func (c KafkaConsumer) Prefix() string
• type Producer
  • func (p Producer) GetClientId() string
  • func (p Producer) GetSecurityProtocol() string
  • func (p Producer) GetTls() *Tls
• type Tls
• type TopicAdmin
  • func NewTopicAdmin(loader config.Loader) (*TopicAdmin, error)
  • func (h TopicAdmin) Prefix() string
• type TopicConsumer

Types

type Admin

type Admin struct {
	BootstrapServers   []string
	ClientId           string
	SecurityProtocol   string
	Tls                *Tls
	CreateTopicTimeout time.Duration `default:"15s"`
}

func (Admin) GetClientId

func (p Admin) GetClientId() string

func (Admin) GetSecurityProtocol

func (p Admin) GetSecurityProtocol() string

func (Admin) GetTls

func (p Admin) GetTls() *Tls

type Client

type Client struct {
	Version          string   `default:"2.1.1"`
	BootstrapServers []string `default:"[\"localhost:9092\"]"`
	SecurityProtocol string   // TLS
	ClientId         string
	Debug            bool
	Tls              *Tls
	Admin            Admin
	Producer         Producer
	Consumer         Consumer
}

func NewClient

func NewClient(loader config.Loader) (*Client, error)

func (*Client) PostBinding

func (p *Client) PostBinding() error

func (Client) Prefix

func (p Client) Prefix() string

type Consumer

type Consumer struct {
	BootstrapServers []string
	ClientId         string
	SecurityProtocol string
	Tls              *Tls
	InitialOffset    int64  `default:"-1"` // -1: Newest, -2: Oldest
	CommitMode       string `default:"AUTO_COMMIT_INTERVAL" validate:"required=false,oneof=AUTO_COMMIT_INTERVAL AUTO_COMMIT_IMMEDIATELY"`
}

func (Consumer) GetClientId

func (p Consumer) GetClientId() string

func (Consumer) GetSecurityProtocol

func (p Consumer) GetSecurityProtocol() string

func (Consumer) GetTls

func (p Consumer) GetTls() *Tls

type EventProducer

type EventProducer struct {
	EventMappings map[string]EventTopic
}

func NewEventProducer

func NewEventProducer(loader config.Loader) (*EventProducer, error)

func (EventProducer) Prefix

func (p EventProducer) Prefix() string

type EventTopic

type EventTopic struct {
	TopicName     string
	Transactional bool `default:"true"`
	Disable       bool
}

type KafkaConsumer

type KafkaConsumer struct {
	HandlerMappings map[string]TopicConsumer
}

func NewKafkaConsumer

func NewKafkaConsumer(loader config.Loader) (*KafkaConsumer, error)

func (KafkaConsumer) Prefix

func (c KafkaConsumer) Prefix() string

type Producer

type Producer struct {
	BootstrapServers []string
	ClientId         string
	SecurityProtocol string
	Tls              *Tls
	FlushMessages    int           `default:"1"`
	FlushFrequency   time.Duration `default:"1s"`
}

func (Producer) GetClientId

func (p Producer) GetClientId() string

func (Producer) GetSecurityProtocol

func (p Producer) GetSecurityProtocol() string

func (Producer) GetTls

func (p Producer) GetTls() *Tls

type Tls

type Tls struct {
	CertFileLocation   string
	KeyFileLocation    string
	CaFileLocation     string
	InsecureSkipVerify bool
}

type TopicAdmin

type TopicAdmin struct {
	Topics []core.TopicConfiguration
}

func NewTopicAdmin

func NewTopicAdmin(loader config.Loader) (*TopicAdmin, error)

func (TopicAdmin) Prefix

func (h TopicAdmin) Prefix() string

type TopicConsumer

type TopicConsumer struct {
	// Enable or disable this consumer
	Enable bool

	// Topic field is high priority than Topics.
	// When Topic is provided, Topics will be ignored.
	Topic string

	// Topics field is lower priority than Topic.
	// When Topic is provided, Topics will be ignored.
	Topics []string

	// GroupId of consumer
	GroupId string

	// TODO implement it
	Concurrency int
}