writer

Documentation

Index

• Constants
• func JSONConversion(a any) (map[string]any, error)
• func StructConversion(a any) (map[string]any, error)
• type Config
  • func (c Config) Validate() error
• type DocumentAdder
• type Request
• type Stats
• type Writer
  • func New(conf Config, client DocumentAdder) (*Writer, error)
  • func (w *Writer) C() chan<- Request
  • func (w *Writer) Run(ctx context.Context)
  • func (w *Writer) Stats() Stats
  • func (w *Writer) Stop()
  • func (w *Writer) Wait()
  • func (w *Writer) Worker(ctx context.Context)
  • func (w *Writer) Workers() int

Constants

const (
	// MaxDocumentCount is the max number of documents that can be sent in one batch
	MaxDocumentCount     = 1_000
	DefaultDocumentCount = MaxDocumentCount
	DefaultFlushInterval = time.Second
)

Functions

func JSONConversion

func JSONConversion(a any) (map[string]any, error)

JSONConversion converts any to a map[string]any using json.Marshal and then json.Unmarshal

func StructConversion

func StructConversion(a any) (map[string]any, error)

StructConversion converts any to a map[string]any using structs.Map, which doesn't honor json tags

Types

type Config

type Config struct {
	// BatchDocumentCount is the max document count to send in one request. Must be less than or equal to MaxDocumentCount
	BatchDocumentCount uint64

	// Workers number of workers to use, defaults to 1.
	Workers uint64
	// FlushInterval is the longest interval to wait before flushing buffered documents.
	FlushInterval time.Duration
	// ConversionFn is a function that converts the data in a Request into a map[string]any.
	// If not set, struct.Map() is used.
	ConversionFn func(any) (map[string]any, error)
}

Config is a struct containing the configurable parameters for a Writer

func (Config) Validate

func (c Config) Validate() error

Validate validates a Config

type DocumentAdder

type DocumentAdder interface {
	AddDocuments(ctx context.Context, workspace, collection string, docs []interface{}) ([]openapi.DocumentStatus, error)
}

DocumentAdder is the interface used to write documents to Rockset, and is implemented by rockset.RockClient.

type Request

type Request struct {
	Workspace  string
	Collection string
	Data       any
}

Request contains a single document to be written to Rockset

type Stats

type Stats struct {
	DocumentCount uint64
	ErrorCount    uint64
}

Stats holds counters for the documents written to Rockset

type Writer

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

Writer is a helper package that writes documents to Rockset collections

func New

func New(conf Config, client DocumentAdder) (*Writer, error)

New creates a new Writer

func (*Writer) C

func (w *Writer) C() chan<- Request

C returns the Request channel, which is used to send documents to be added to Rockset.

func (*Writer) Run

func (w *Writer) Run(ctx context.Context)

Run starts the reader loop that gets write requests from the channel and batches them so the workers can add them to the collection(s). It starts the number Worker specified in the Config, but more can be started it if the number of collections written to is large.

rs, _ := rockset.NewClient() w := writer.New(Config{}, rs.Documents) go w.Run(ctx) ... w.Stop() w.Wait()

func (*Writer) Stats

func (w *Writer) Stats() Stats

Stats returns a struct with document write statistics

func (*Writer) Stop

func (w *Writer) Stop()

Stop cleanly stops the Writer and flushes any buffered item, and closes the Request channel.

func (*Writer) Wait

func (w *Writer) Wait()

Wait waits until the reader loop and all workers have finished

func (*Writer) Worker

func (w *Writer) Worker(ctx context.Context)

Worker runs a worker that writes batches of documents to the Rockset API. It needs to be started as a go routine, or it will block.

func (*Writer) Workers

func (w *Writer) Workers() int

Workers returns the number of workers