kafka

Documentation

Index

• type Configuration
  • func NewConfiguration(id string, triggerConfiguration *functionconfig.Trigger, ...) (*Configuration, error)
• type Event
  • func (e *Event) GetBody() []byte
  • func (e *Event) GetHeader(key string) interface{}
  • func (e *Event) GetHeaders() map[string]interface{}
  • func (e *Event) GetOffset() int
  • func (e *Event) GetPath() string
  • func (e *Event) GetShardID() int
  • func (e *Event) GetSize() int
  • func (e *Event) GetTimestamp() time.Time
  • func (e *Event) GetTopic() string
• type SaramaLogger
  • func NewSaramaLogger(parentLogger logger.Logger) SaramaLogger
  • func (s SaramaLogger) Print(v ...interface{})
  • func (s SaramaLogger) Printf(format string, v ...interface{})
  • func (s SaramaLogger) Println(v ...interface{})

Types

type Configuration

type Configuration struct {
	trigger.Configuration

	Brokers         []string
	Topics          []string
	ConsumerGroup   string
	InitialOffset   string
	BalanceStrategy string
	SASL            struct {
		Enable    bool
		Handshake *bool
		User      string
		Password  string
		Mechanism string

		// oauth
		OAuth struct {
			ClientID     string
			ClientSecret string
			TokenURL     string
			Scopes       []string
		}
	}

	TLS struct {
		Enable             bool
		InsecureSkipVerify bool
		MinimumVersion     string
	}

	SecretPath string

	SessionTimeout                string
	HeartbeatInterval             string
	MaxProcessingTime             string
	RebalanceTimeout              string
	RebalanceRetryBackoff         string
	RetryBackoff                  string
	MaxWaitTime                   string
	MaxWaitHandlerDuringRebalance string
	WorkerAllocationMode          partitionworker.AllocationMode
	RebalanceRetryMax             int
	FetchMin                      int
	FetchDefault                  int
	FetchMax                      int
	ChannelBufferSize             int
	CACert                        string
	AccessKey                     string
	AccessCertificate             string
	LogLevel                      int
	AckWindowSize                 int
	Version                       string
	// contains filtered or unexported fields
}

func NewConfiguration

func NewConfiguration(id string,
	triggerConfiguration *functionconfig.Trigger,
	runtimeConfiguration *runtime.Configuration,
	logger logger.Logger) (*Configuration, error)

type Event

type Event struct {
	nuclio.AbstractEvent
	// contains filtered or unexported fields
}

func (*Event) GetBody

func (e *Event) GetBody() []byte

func (*Event) GetHeader

func (e *Event) GetHeader(key string) interface{}

GetHeader returns the header by name as an interface{}

func (*Event) GetHeaders

func (e *Event) GetHeaders() map[string]interface{}

func (*Event) GetOffset

func (e *Event) GetOffset() int

func (*Event) GetPath

func (e *Event) GetPath() string

func (*Event) GetShardID

func (e *Event) GetShardID() int

func (*Event) GetSize

func (e *Event) GetSize() int

func (*Event) GetTimestamp

func (e *Event) GetTimestamp() time.Time

func (*Event) GetTopic

func (e *Event) GetTopic() string

type SaramaLogger

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

func NewSaramaLogger

func NewSaramaLogger(parentLogger logger.Logger) SaramaLogger

func (SaramaLogger) Print

func (s SaramaLogger) Print(v ...interface{})

func (SaramaLogger) Printf

func (s SaramaLogger) Printf(format string, v ...interface{})

func (SaramaLogger) Println

func (s SaramaLogger) Println(v ...interface{})