messager

Documentation

Index

• Variables
• type Engine
  • func NewEngine(tsv TabletService, se *schema.Engine, vs VStreamer) *Engine
  • func (me *Engine) Close()
  • func (me *Engine) GenerateAckQuery(name string, ids []string) (string, map[string]*querypb.BindVariable, error)
  • func (me *Engine) GeneratePostponeQuery(name string, ids []string) (string, map[string]*querypb.BindVariable, error)
  • func (me *Engine) GeneratePurgeQuery(name string, timeCutoff int64) (string, map[string]*querypb.BindVariable, error)
  • func (me *Engine) Open()
  • func (me *Engine) Subscribe(ctx context.Context, name string, send func(*sqltypes.Result) error) (done <-chan struct{}, err error)
• type MessageRow
  • func BuildMessageRow(row []sqltypes.Value) (*MessageRow, error)
• type TabletService
• type VStreamer

Variables

var (
	// MessageStats tracks stats for messages.
	MessageStats = stats.NewGaugesWithMultiLabels(
		"Messages",
		"Stats for messages",
		[]string{"TableName", "Metric"})
)

Types

type Engine

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

Engine is the engine for handling messages.

func NewEngine

func NewEngine(tsv TabletService, se *schema.Engine, vs VStreamer) *Engine

NewEngine creates a new Engine.

func (*Engine) Close

func (me *Engine) Close()

Close closes the Engine service.

func (*Engine) GenerateAckQuery

func (me *Engine) GenerateAckQuery(name string, ids []string) (string, map[string]*querypb.BindVariable, error)

GenerateAckQuery returns the query and bind vars for acking a message.

func (*Engine) GeneratePostponeQuery

func (me *Engine) GeneratePostponeQuery(name string, ids []string) (string, map[string]*querypb.BindVariable, error)

GeneratePostponeQuery returns the query and bind vars for postponing a message.

func (*Engine) GeneratePurgeQuery

func (me *Engine) GeneratePurgeQuery(name string, timeCutoff int64) (string, map[string]*querypb.BindVariable, error)

GeneratePurgeQuery returns the query and bind vars for purging messages.

func (*Engine) Open

func (me *Engine) Open()

Open starts the Engine service.

func (*Engine) Subscribe

func (me *Engine) Subscribe(ctx context.Context, name string, send func(*sqltypes.Result) error) (done <-chan struct{}, err error)

Subscribe subscribes to messages from the requested table. The function returns a done channel that will be closed when the subscription ends, which can be initiated by the send function returning io.EOF. The engine can also end a subscription which is usually triggered by Close. It's the responsibility of the send function to promptly return if the done channel is closed. Otherwise, the engine's Close function will hang indefinitely.

type MessageRow

type MessageRow struct {
	Priority  int64
	TimeNext  int64
	Epoch     int64
	TimeAcked int64
	Row       []sqltypes.Value
	// contains filtered or unexported fields
}

MessageRow represents a message row. The first column in Row is always the "id".

func BuildMessageRow

func BuildMessageRow(row []sqltypes.Value) (*MessageRow, error)

BuildMessageRow builds a MessageRow for a db row.

type TabletService

type TabletService interface {
	tabletenv.Env
	PostponeMessages(ctx context.Context, target *querypb.Target, name string, ids []string) (count int64, err error)
	PurgeMessages(ctx context.Context, target *querypb.Target, name string, timeCutoff int64) (count int64, err error)
}

TabletService defines the functions of TabletServer that the messager needs for callback.

type VStreamer

type VStreamer interface {
	Stream(ctx context.Context, startPos string, tablePKs []*binlogdatapb.TableLastPK, filter *binlogdatapb.Filter, send func([]*binlogdatapb.VEvent) error) error
	StreamResults(ctx context.Context, query string, send func(*binlogdatapb.VStreamResultsResponse) error) error
}

VStreamer defines the functions of VStreamer that the messager needs.