io

Documentation

Index

• func Batch(ch <-chan interface{}, c *Config) (batch []interface{}, ident string)
• func BatchNumber() string
• func LoadConfig(b json.RawMessage, i interface{}) error
• func ReadConfig(f *os.File) (json.RawMessage, error)
• func StartHeartbeat(version, uid, hook string, interval, timeout int) error
• func Validator() *validator.Validate
• type Config
  • func (c Config) Log(msg string, args ...interface{})
• type FileSink
  • func (f *FileSink) Flush(uuid, filename string, d []byte) error
  • func (f *FileSink) LoadConfig(b json.RawMessage) error
• type Flusher
  • func GetFlusher(conf *Config) (Flusher, error)
• type MemSink
  • func (m *MemSink) Flush(uuid, ident string, d []byte) error
  • func (m *MemSink) LoadConfig(_ json.RawMessage) error
• type S3Sink
  • func (s *S3Sink) Flush(uuid, filename string, d []byte) error
  • func (s *S3Sink) LoadConfig(b json.RawMessage) error

Functions

func Batch

func Batch(ch <-chan interface{}, c *Config) (batch []interface{}, ident string)

Listen to an Interface channel and return a buffer batch on Either a timeout happens OR buffer is filled to a size.

func BatchNumber

func BatchNumber() string

func LoadConfig

func LoadConfig(b json.RawMessage, i interface{}) error

func ReadConfig

func ReadConfig(f *os.File) (json.RawMessage, error)

func StartHeartbeat

func StartHeartbeat(version, uid, hook string, interval, timeout int) error

func Validator

func Validator() *validator.Validate

Types

type Config

type Config struct {
	Debug             bool            `json:"debug"`
	UID               string          `json:"uid" validate:"required"`
	BatchSize         int             `json:"batch_size"`
	BatchInterval     int             `json:"batch_interval"`
	Sink              string          `json:"sink" validate:"required"`
	Raw               json.RawMessage `json:"-"`
	HeartbeatHook     string          `json:"heartbeat_hook"`
	HeartbeatInterval int             `json:"heartbeat_interval"`
	HeartbeatTimeout  int             `json:"heartbeat_timeout_ms"`
}

func (Config) Log

func (c Config) Log(msg string, args ...interface{})

type FileSink

type FileSink struct {
	Dir string `json:"file_sink_dir" validate:"required"`
}

func (*FileSink) Flush

func (f *FileSink) Flush(uuid, filename string, d []byte) error

func (*FileSink) LoadConfig

func (f *FileSink) LoadConfig(b json.RawMessage) error

type Flusher

type Flusher interface {
	Flush(uuid, ident string, d []byte) error
	LoadConfig(json.RawMessage) error
}

func GetFlusher

func GetFlusher(conf *Config) (Flusher, error)

type MemSink

type MemSink struct {
	Records map[string][]byte
	OnFetch func(string)
	// contains filtered or unexported fields
}

func (*MemSink) Flush

func (m *MemSink) Flush(uuid, ident string, d []byte) error

func (*MemSink) LoadConfig

func (m *MemSink) LoadConfig(_ json.RawMessage) error

type S3Sink

type S3Sink struct {
	Prefix  string `json:"prefix" validate:"required"`
	Region  string `json:"aws_region", validate:"required"`
	Bucket  string `json:"aws_bucket" validate:"required"`
	Profile string `json:"aws_profile" validate:"required"`
}

func (*S3Sink) Flush

func (s *S3Sink) Flush(uuid, filename string, d []byte) error

func (*S3Sink) LoadConfig

func (s *S3Sink) LoadConfig(b json.RawMessage) error