gears

Documentation

Index

• Variables
• func NewManualHashPartitioner(topic string) sarama.Partitioner
• func NewPlugin() (*pkgplugin.Plugin, error)
• func NewPluginVersion(name string, version string, commitID string) (*pkgplugin.Plugin, error)
• func NewSender(tid tenant.Id, plugin string, name string, config interface{}, ...) (sender.Sender, error)
• type DynamicMetricLabel
• type DynamicMetricValue
• type ManualHashPartitioner
  • func (mp *ManualHashPartitioner) Partition(message *sarama.ProducerMessage, numPartitions int32) (int32, error)
• type Producer
  • func (p *Producer) Close(ctx context.Context)
  • func (p *Producer) Partitions(topic string) ([]int32, error)
  • func (p *Producer) SendMessage(ctx context.Context, topic string, partition int, headers map[string]string, ...) error
• type Sender
  • func (s *Sender) Config() interface{}
  • func (s *Sender) Count() int
  • func (s *Sender) EventErrorCount() int
  • func (s *Sender) EventErrorVelocity() int
  • func (s *Sender) EventSuccessCount() int
  • func (s *Sender) EventSuccessVelocity() int
  • func (s *Sender) EventTs() int64
  • func (s *Sender) Hash() string
  • func (s *Sender) Name() string
  • func (s *Sender) NewProducer(count int) (*Producer, error)
  • func (s *Sender) NewSyncProducers(count int) ([]sarama.SyncProducer, sarama.Client, error)
  • func (s *Sender) Plugin() string
  • func (s *Sender) Send(e event.Event)
  • func (s *Sender) StopSending(ctx context.Context)
  • func (s *Sender) Tenant() tenant.Id
  • func (s *Sender) Unwrap() sender.Sender
• type SenderConfig
  • func (sc *SenderConfig) Validate() error
  • func (sc SenderConfig) WithDefaults() SenderConfig
• type SenderMetrics

Variables

var (
	Name     = "gears"
	Version  = "v0.0.0"
	CommitID = ""
)

var DefaultSenderConfig = SenderConfig{
	Brokers:             "localhost:9092",
	Topic:               "",
	ChannelBufferSize:   pointer.Int(0),
	Username:            "",
	Password:            "",
	CACert:              "",
	AccessCert:          "",
	AccessKey:           "",
	Version:             "",
	SenderPoolSize:      pointer.Int(1),
	DynamicMetricLabels: make([]DynamicMetricLabel, 0),
	Location:            "",
	App:                 "",
	Partner:             "",
	Uses:                "",
}

Functions

func NewManualHashPartitioner

func NewManualHashPartitioner(topic string) sarama.Partitioner

func NewPlugin

func NewPlugin() (*pkgplugin.Plugin, error)

func NewPluginVersion

func NewPluginVersion(name string, version string, commitID string) (*pkgplugin.Plugin, error)

func NewSender

func NewSender(tid tenant.Id, plugin string, name string, config interface{}, secrets secret.Vault, tableSyncer syncer.DeltaSyncer) (sender.Sender, error)

Types

type DynamicMetricLabel

type DynamicMetricLabel struct {
	Label string `json:"label,omitempty"`
	Path  string `json:"path,omitempty"`
}

type DynamicMetricValue

type DynamicMetricValue struct {
	Label string `json:"label,omitempty"`
	Value string `json:"value,omitempty"`
}

type ManualHashPartitioner

type ManualHashPartitioner struct {
	sarama.Partitioner
}

func (*ManualHashPartitioner) Partition

func (mp *ManualHashPartitioner) Partition(message *sarama.ProducerMessage, numPartitions int32) (int32, error)

type Producer

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

func (*Producer) Close

func (p *Producer) Close(ctx context.Context)

func (*Producer) Partitions

func (p *Producer) Partitions(topic string) ([]int32, error)

func (*Producer) SendMessage

func (p *Producer) SendMessage(ctx context.Context, topic string, partition int, headers map[string]string, bs []byte, e event.Event) error

type Sender

type Sender struct {
	sync.Mutex
	// contains filtered or unexported fields
}

func (*Sender) Config

func (s *Sender) Config() interface{}

func (*Sender) Count

func (s *Sender) Count() int

func (*Sender) EventErrorCount

func (s *Sender) EventErrorCount() int

func (*Sender) EventErrorVelocity

func (s *Sender) EventErrorVelocity() int

func (*Sender) EventSuccessCount

func (s *Sender) EventSuccessCount() int

func (*Sender) EventSuccessVelocity

func (s *Sender) EventSuccessVelocity() int

func (*Sender) EventTs

func (s *Sender) EventTs() int64

func (*Sender) Hash

func (s *Sender) Hash() string

func (*Sender) Name

func (s *Sender) Name() string

func (*Sender) NewProducer

func (s *Sender) NewProducer(count int) (*Producer, error)

func (*Sender) NewSyncProducers

func (s *Sender) NewSyncProducers(count int) ([]sarama.SyncProducer, sarama.Client, error)

func (*Sender) Plugin

func (s *Sender) Plugin() string

func (*Sender) Send

func (s *Sender) Send(e event.Event)

func (*Sender) StopSending

func (s *Sender) StopSending(ctx context.Context)

func (*Sender) Tenant

func (s *Sender) Tenant() tenant.Id

func (*Sender) Unwrap

func (s *Sender) Unwrap() sender.Sender

type SenderConfig

type SenderConfig struct {
	Brokers             string               `json:"brokers,omitempty"`
	Topic               string               `json:"topic,omitempty"`
	Username            string               `json:"username,omitempty"`
	Password            string               `json:"password,omitempty"`
	CACert              string               `json:"caCert,omitempty"`
	AccessCert          string               `json:"accessCert,omitempty"`
	AccessKey           string               `json:"accessKey,omitempty"`
	Version             string               `json:"version,omitempty"`
	ChannelBufferSize   *int                 `json:"channelBufferSize,omitempty"`
	TLSEnable           bool                 `json:"tlsEnable,omitempty"`
	SenderPoolSize      *int                 `json:"senderPoolSize,omitempty"`
	DynamicMetricLabels []DynamicMetricLabel `json:"dynamicMetricLabel,omitempty"`
	CompressionMethod   string               `json:"compressionMethod,omitempty"`
	CompressionLevel    *int                 `json:"compressionLevel,omitempty"`
	Location            interface{}          `json:"location,omitempty"` // gears config: string or array of strings, may contain path
	App                 string               `json:"app,omitempty"`      // gears config
	Partner             string               `json:"partner,omitempty"`  // gears config
	Uses                string               `json:"uses,omitempty"`     // gears config
}

SenderConfig can be passed into NewSender() in order to configure the behavior of the sender.

func (*SenderConfig) Validate

func (sc *SenderConfig) Validate() error

Validate

func (SenderConfig) WithDefaults

func (sc SenderConfig) WithDefaults() SenderConfig

WithDefaults

type SenderMetrics

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