transport

Documentation

Index

• Variables
• func HashProto(fields []string, flowMessage *flowmessage.FlowMessage) string
• func ParseKafkaVersion(versionString string) (sarama.KafkaVersion, error)
• func RegisterFlags()
• func SetKafkaVersion(version sarama.KafkaVersion)
• type KafkaState
  • func StartKafkaProducer(addrs []string, topic string, hashing bool, keying string, useTls bool, ...) (*KafkaState, error)
  • func StartKafkaProducerFromArgs(log utils.Logger) (*KafkaState, error)
  • func (s KafkaState) Publish(msgs []*flowmessage.FlowMessage)
  • func (s KafkaState) SendKafkaFlowMessage(flowMessage *flowmessage.FlowMessage)

Variables

var (
	KafkaTLS   *bool
	KafkaSASL  *bool
	KafkaTopic *string
	KafkaSrv   *string
	KafkaBrk   *string

	KafkaLogErrors *bool

	KafkaHashing *bool
	KafkaKeying  *string
	KafkaVersion *string
)

Functions

func HashProto

func HashProto(fields []string, flowMessage *flowmessage.FlowMessage) string

func ParseKafkaVersion

func ParseKafkaVersion(versionString string) (sarama.KafkaVersion, error)

ParseKafkaVersion is a pass through to sarama.ParseKafkaVersion to get a KafkaVersion struct by a string version that can be passed into SetKafkaVersion This function is here so that calling code need not import sarama to set KafkaVersion

func RegisterFlags

func RegisterFlags()

func SetKafkaVersion

func SetKafkaVersion(version sarama.KafkaVersion)

SetKafkaVersion sets the KafkaVersion that is used to set the log message format version

Types

type KafkaState

type KafkaState struct {
	FixedLengthProto bool
	// contains filtered or unexported fields
}

func StartKafkaProducer

func StartKafkaProducer(addrs []string, topic string, hashing bool, keying string, useTls bool, useSasl bool, logErrors bool, log utils.Logger) (*KafkaState, error)

func StartKafkaProducerFromArgs

func StartKafkaProducerFromArgs(log utils.Logger) (*KafkaState, error)

func (KafkaState) Publish

func (s KafkaState) Publish(msgs []*flowmessage.FlowMessage)

func (KafkaState) SendKafkaFlowMessage

func (s KafkaState) SendKafkaFlowMessage(flowMessage *flowmessage.FlowMessage)