kafka

Documentation

Index

• Variables
• func CloseProducer(producer interface{}) error
• func GetStatusCodeFromError(err error) int
• func NewProducer(destinationConfig interface{}) (sarama.SyncProducer, error)
• func NewProducerForAzureEventHub(destinationConfig interface{}) (sarama.SyncProducer, error)
• func NewProducerForConfluentCloud(destinationConfig interface{}) (sarama.SyncProducer, error)
• func NewTLSConfig(caCertFile string) *tls.Config
• func Produce(jsonData json.RawMessage, producer interface{}, destConfig interface{}) (int, string, string)
• func SetSASLConfig(config *sarama.Config, destConfig Config) (err error)
• type AzureEventHubConfig
• type Config
• type ConfluentCloudConfig
• type XDGSCRAMClient
  • func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)
  • func (x *XDGSCRAMClient) Done() bool
  • func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)

Variables

var (
	SHA256 scram.HashGeneratorFcn = sha256.New
	SHA512 scram.HashGeneratorFcn = sha512.New
)

Functions

func CloseProducer

func CloseProducer(producer interface{}) error

CloseProducer closes a given producer

func GetStatusCodeFromError

func GetStatusCodeFromError(err error) int

GetStatusCodeFromError parses the error and returns the status so that event gets retried or failed.

func NewProducer

func NewProducer(destinationConfig interface{}) (sarama.SyncProducer, error)

NewProducer creates a producer based on destination config

func NewProducerForAzureEventHub

func NewProducerForAzureEventHub(destinationConfig interface{}) (sarama.SyncProducer, error)

NewProducerForAzureEventHub creates a producer for Azure event hub based on destination config

func NewProducerForConfluentCloud

func NewProducerForConfluentCloud(destinationConfig interface{}) (sarama.SyncProducer, error)

NewProducerForConfluentCloud creates a producer for Confluent cloud based on destination config

func NewTLSConfig

func NewTLSConfig(caCertFile string) *tls.Config

NewTLSConfig generates a TLS configuration used to authenticate on server with certificates.

func Produce

func Produce(jsonData json.RawMessage, producer interface{}, destConfig interface{}) (int, string, string)

Produce creates a producer and send data to Kafka.

func SetSASLConfig

func SetSASLConfig(config *sarama.Config, destConfig Config) (err error)

Sets SASL authentication config for Kafka

Types

type AzureEventHubConfig

type AzureEventHubConfig struct {
	Topic                     string
	BootstrapServer           string
	EventHubsConnectionString string
}

AzureEventHubConfig is the config that is required to send data to Azure Event Hub

type Config

type Config struct {
	Topic         string
	HostName      string
	Port          string
	SslEnabled    bool
	CACertificate string
	UseSASL       bool
	SaslType      string
	Username      string
	Password      string
}

Config is the config that is required to send data to Kafka

type ConfluentCloudConfig

type ConfluentCloudConfig struct {
	Topic           string
	BootstrapServer string
	APIKey          string
	APISecret       string
}

ConfluentCloudConfig is the config that is required to send data to Confluent Cloud

type XDGSCRAMClient

type XDGSCRAMClient struct {
	*scram.Client
	*scram.ClientConversation
	scram.HashGeneratorFcn
}

Boilerplate needed for SCRAM Authentication in Kafka

func (*XDGSCRAMClient) Begin

func (x *XDGSCRAMClient) Begin(userName, password, authzID string) (err error)

func (*XDGSCRAMClient) Done

func (x *XDGSCRAMClient) Done() bool

func (*XDGSCRAMClient) Step

func (x *XDGSCRAMClient) Step(challenge string) (response string, err error)