http

Documentation

Index

• Constants
• type CompressionStrategy
• type Compressor
  • func (c *Compressor) AddHeaders(req *http.Request)
  • func (c *Compressor) Compress(in *bytes.Buffer) (*bytes.Buffer, error)
• type Config
  • func (c *Config) Validate() error
• type HTTP
  • func New(name string, config *Config, log logrus.FieldLogger, ...) (*HTTP, error)
  • func (h *HTTP) HandleNewDecoratedEvent(ctx context.Context, event *xatu.DecoratedEvent) error
  • func (h *HTTP) HandleNewDecoratedEvents(ctx context.Context, events []*xatu.DecoratedEvent) error
  • func (h *HTTP) Name() string
  • func (h *HTTP) Start(ctx context.Context) error
  • func (h *HTTP) Stop(ctx context.Context) error
  • func (h *HTTP) Type() string
• type ItemExporter
  • func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemExporter, error)
  • func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEvent) error
  • func (e ItemExporter) Shutdown(ctx context.Context) error

Constants

const SinkType = "http"

Types

type CompressionStrategy

type CompressionStrategy string

var (
	CompressionStrategyNone   CompressionStrategy = "none"
	CompressionStrategyGzip   CompressionStrategy = "gzip"
	CompressionStrategyZstd   CompressionStrategy = "zstd"
	CompressionStrategyZlib   CompressionStrategy = "zlib"
	CompressionStrategySnappy CompressionStrategy = "snappy"
)

type Compressor

type Compressor struct {
	Strategy CompressionStrategy
}

func (*Compressor) AddHeaders

func (c *Compressor) AddHeaders(req *http.Request)

func (*Compressor) Compress

func (c *Compressor) Compress(in *bytes.Buffer) (*bytes.Buffer, error)

type Config

type Config struct {
	Address            string              `yaml:"address"`
	Headers            map[string]string   `yaml:"headers"`
	MaxQueueSize       int                 `yaml:"maxQueueSize" default:"51200"`
	BatchTimeout       time.Duration       `yaml:"batchTimeout" default:"5s"`
	ExportTimeout      time.Duration       `yaml:"exportTimeout" default:"30s"`
	MaxExportBatchSize int                 `yaml:"maxExportBatchSize" default:"512"`
	Compression        CompressionStrategy `yaml:"compression" default:"none"`
	KeepAlive          *bool               `yaml:"keepAlive" default:"true"`
	Workers            int                 `yaml:"workers" default:"1"`
}

func (*Config) Validate

func (c *Config) Validate() error

type HTTP

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

func New

func New(name string, config *Config, log logrus.FieldLogger, filterConfig *xatu.EventFilterConfig, shippingMethod processor.ShippingMethod) (*HTTP, error)

func (*HTTP) HandleNewDecoratedEvent

func (h *HTTP) HandleNewDecoratedEvent(ctx context.Context, event *xatu.DecoratedEvent) error

func (*HTTP) HandleNewDecoratedEvents

func (h *HTTP) HandleNewDecoratedEvents(ctx context.Context, events []*xatu.DecoratedEvent) error

func (*HTTP) Name

func (h *HTTP) Name() string

func (*HTTP) Start

func (h *HTTP) Start(ctx context.Context) error

func (*HTTP) Stop

func (h *HTTP) Stop(ctx context.Context) error

func (*HTTP) Type

func (h *HTTP) Type() string

type ItemExporter

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

func NewItemExporter

func NewItemExporter(name string, config *Config, log logrus.FieldLogger) (ItemExporter, error)

func (ItemExporter) ExportItems

func (e ItemExporter) ExportItems(ctx context.Context, items []*xatu.DecoratedEvent) error

func (ItemExporter) Shutdown

func (e ItemExporter) Shutdown(ctx context.Context) error