Documentation
¶
Overview ¶
Package streamingbatchwriter provides a writers.Writer implementation that writes to a client that implements the streamingbatchwriter.Client interface.
Write messages are sent to the client with three separate methods: MigrateTable, WriteTable, and DeleteStale. Each method is called separate goroutines. Message types are processed in blocks: Receipt of a new message type will cause the previous message type processing to end (if it exists) which is signalled to the handler by closing the channel. The handler should return after processing all messages.
For Insert messages (handled by WriteTable) each table creates separate goroutine. Number of goroutines is limited by the number of tables. Thus, each WriteTable invocation is for a single table (all messages sent to WriteTable are guaranteed to be for the same table).
After a 'batch' is complete, the channel is closed. The handler is expected to block until the channel is closed and to keep processing in a streaming fashion.
Batches are considered complete when: 1. The batch timeout is reached 2. The batch size is reached 3. The batch size in bytes is reached 4. A different message type is received
Each handler can get invoked multiple times as new batches are processed. Handlers get invoked only if there's a message of that type at hand: First message of the batch is immediately available in the channel.
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
func DummyHandler ¶
func DummyHandler[T message.WriteMessage](ch <-chan T)
DummyHandler should be used to empty Migration and DeleteStale channels if they are not used.
Types ¶
type Client ¶
type Client interface {
// MigrateTable should block and handle WriteMigrateTable messages until the channel is closed.
MigrateTable(context.Context, <-chan *message.WriteMigrateTable) error
// DeleteStale should block and handle WriteDeleteStale messages until the channel is closed.
DeleteStale(context.Context, <-chan *message.WriteDeleteStale) error
// WriteTable should block and handle writes to a single table until the channel is closed. Table metadata can be found in the first WriteInsert message.
// The channel is closed when all inserts in the batch have been sent. New batches, if any, will be sent on a new call to WriteTable.
WriteTable(context.Context, <-chan *message.WriteInsert) error
}
Client is the interface that must be implemented by the client of StreamingBatchWriter.
type Option ¶
type Option func(*StreamingBatchWriter)
func WithBatchSizeBytes ¶
func WithBatchSizeRows ¶
func WithBatchTimeout ¶
func WithLogger ¶
type StreamingBatchWriter ¶
type StreamingBatchWriter struct {
// contains filtered or unexported fields
}
func (*StreamingBatchWriter) Write ¶
func (w *StreamingBatchWriter) Write(ctx context.Context, msgs <-chan message.WriteMessage) error
Source Files