vstreamer

Documentation

Index

• Variables
• type ColExpr
• type Engine
  • func NewEngine(ts srvtopo.Server, se *schema.Engine) *Engine
  • func (vse *Engine) Close()
  • func (vse *Engine) InitDBConfig(cp dbconfigs.Connector)
  • func (vse *Engine) IsOpen() bool
  • func (vse *Engine) Open(keyspace, cell string) error
  • func (vse *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request)
  • func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogdatapb.Filter, ...) error
  • func (vse *Engine) StreamResults(ctx context.Context, query string, ...) error
  • func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, ...) error
• type Plan
• type RowStreamer
  • func NewRowStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, query string, ...) RowStreamer
• type Table
• type VStreamer
  • func NewVStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, ...) VStreamer

Variables

var HeartbeatTime = 900 * time.Millisecond

HeartbeatTime is set to slightly below 1s, compared to idleTimeout set by VPlayer at slightly above 1s. This minimizes conflicts between the two timeouts.

var PacketSize = flag.Int("vstream_packet_size", 30000, "Suggested packet size for VReplication streamer. This is used only as a recommendation. The actual packet size may be more or less than this amount.")

PacketSize is the suggested packet size for VReplication streamer.

Types

type ColExpr

type ColExpr struct {
	// ColNum specifies the source column value.
	ColNum int

	// Vindex and VindexColumns, if set, will be used to generate
	// a keyspace_id. If so, ColNum is ignored.
	// VindexColumns contains the column numbers of the table,
	// and not the column numbers of the stream to be sent.
	Vindex        vindexes.Vindex
	VindexColumns []int

	// Alias is usually the column name, but it can be changed
	// if the select expression aliases with an "AS" expression.
	// Also, "keyspace_id()" will be aliased as "keyspace_id".
	// This Alias is sent as field info for the returned stream.
	Alias sqlparser.ColIdent
	Type  querypb.Type
}

ColExpr represents a column expression.

type Engine

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

Engine is the engine for handling vreplication streaming requests.

func NewEngine

func NewEngine(ts srvtopo.Server, se *schema.Engine) *Engine

NewEngine creates a new Engine. Initialization sequence is: NewEngine->InitDBConfig->Open. Open and Close can be called multiple times and are idempotent.

func (*Engine) Close

func (vse *Engine) Close()

Close closes the Engine service.

func (*Engine) InitDBConfig

func (vse *Engine) InitDBConfig(cp dbconfigs.Connector)

InitDBConfig performs saves the required info from dbconfigs for future use.

func (*Engine) IsOpen

func (vse *Engine) IsOpen() bool

IsOpen checks if the engine is opened

func (*Engine) Open

func (vse *Engine) Open(keyspace, cell string) error

Open starts the Engine service.

func (*Engine) ServeHTTP

func (vse *Engine) ServeHTTP(response http.ResponseWriter, request *http.Request)

ServeHTTP shows the current VSchema.

func (*Engine) Stream

func (vse *Engine) Stream(ctx context.Context, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error

Stream starts a new stream.

func (*Engine) StreamResults

func (vse *Engine) StreamResults(ctx context.Context, query string, send func(*binlogdatapb.VStreamResultsResponse) error) error

StreamResults streams results of the query with the gtid.

func (*Engine) StreamRows

func (vse *Engine) StreamRows(ctx context.Context, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) error) error

StreamRows streams rows.

type Plan

type Plan struct {
	Table *Table
	// ColExprs is the list of column expressions to be sent
	// in the stream.
	ColExprs []ColExpr

	// Vindex, VindexColumns and KeyRange, if set, will be used
	// to filter the row.
	// VindexColumns contains the column numbers of the table,
	// and not the column numbers of the stream to be sent.
	Vindex        vindexes.Vindex
	VindexColumns []int
	KeyRange      *topodatapb.KeyRange
}

Plan represents the plan for a table.

type RowStreamer

type RowStreamer interface {
	Stream() error
	Cancel()
}

RowStreamer exposes an externally usable interface to rowStreamer.

func NewRowStreamer

func NewRowStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, query string, lastpk []sqltypes.Value, send func(*binlogdatapb.VStreamRowsResponse) error) RowStreamer

NewRowStreamer returns a RowStreamer

type Table

type Table struct {
	Name   string
	Fields []*querypb.Field
}

Table contains the metadata for a table.

type VStreamer

type VStreamer interface {
	Stream() error
	Cancel()
}

VStreamer exposes an externally usable interface to vstreamer.

func NewVStreamer

func NewVStreamer(ctx context.Context, cp dbconfigs.Connector, se *schema.Engine, startPos string, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) VStreamer

NewVStreamer returns a VStreamer.