search

Documentation

Index

• Variables
• type BulkElastic
  • func NewElastic(opts BulkElasticOpts) (es *BulkElastic, err error)
  • func (e *BulkElastic) Add(pos pglogrepl.LSN, buffers ...[]byte) error
  • func (e *BulkElastic) PrepareScripts() error
  • func (e *BulkElastic) Start(wg *sync.WaitGroup, ctx context.Context)
• type BulkElasticOpts
• type BulkResponseErrors
  • func (bs *BulkResponseErrors) UnmarshalJSON(b []byte) error
• type BulkRowError
  • func (err BulkRowError) Error() string
• type Client
  • func NewClient(host, username, password string, logger *zap.Logger) (c *Client, err error)
  • func (c *Client) Bulk(body io.Reader) error
  • func (c *Client) Do(req *http.Request) (*http.Response, error)
  • func (c *Client) Script(id, source string) error
• type Credentials
• type ErrHTTP
  • func (e ErrHTTP) Error() string

Variables

var (
	ErrBulkCommitFail = errors.New("commit bulk returned errors")
)

var (
	ErrEmptyDocumentStream = errors.New("document stream can not be empty")
)

Types

type BulkElastic

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

func NewElastic

func NewElastic(opts BulkElasticOpts) (es *BulkElastic, err error)

func (*BulkElastic) Add

func (e *BulkElastic) Add(pos pglogrepl.LSN, buffers ...[]byte) error

func (*BulkElastic) PrepareScripts

func (e *BulkElastic) PrepareScripts() error

func (*BulkElastic) Start

func (e *BulkElastic) Start(wg *sync.WaitGroup, ctx context.Context)

type BulkElasticOpts

type BulkElasticOpts struct {
	Host     string
	Username string
	Password string

	//
	Logger *zap.Logger
	// Maximum waiting time for data. Any partial bulk request will be pushed after idleInterval.
	// Default: 5s
	IdleInterval time.Duration
	// Minimal time between requests.
	// Default: 500ms
	Throttle time.Duration
	// Time after request, if
	Debounce time.Duration
	// Maximum bulk request
	// Good one would be 4-8mb; limit ~100MBsize in megabytes
	// Default: 4M
	BulkSize int
	// document stream
	Stream *postgres.StreamPipe
}

type BulkResponseErrors

type BulkResponseErrors struct {
	Errors []BulkRowError
}

func (*BulkResponseErrors) UnmarshalJSON

func (bs *BulkResponseErrors) UnmarshalJSON(b []byte) error

type BulkRowError

type BulkRowError struct {
	DocID  string
	Type   string `json:"type"`
	Reason string `json:"reason"`
}

func (BulkRowError) Error

func (err BulkRowError) Error() string

type Client

type Client struct {
	http.Client

	Host *url.URL
	// contains filtered or unexported fields
}

func NewClient

func NewClient(host, username, password string, logger *zap.Logger) (c *Client, err error)

func (*Client) Bulk

func (c *Client) Bulk(body io.Reader) error

Bulk request with basic error handling XXX: When using the HTTP API, make sure that the client does not send HTTP chunks, as this will slow things down. See: https://www.elastic.co/guide/en/elasticsearch/reference/7.10/docs-bulk.html

func (*Client) Do

func (c *Client) Do(req *http.Request) (*http.Response, error)

Do wraps default http.Client.Do with authorization

func (*Client) Script

func (c *Client) Script(id, source string) error

type Credentials

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

type ErrHTTP

type ErrHTTP struct {
	StatusCode int
}

ErrHTTP is a wrapper on http status codes.

func (ErrHTTP) Error

func (e ErrHTTP) Error() string