replication

Documentation

Index

• Variables
• func WTSetToCSM(wtSet *wal.WTSet) (io.ColumnSeriesMap, error)
• type GRPCClient
• type GRPCReplicationClient
  • func NewGRPCReplicationClient(client pb.ReplicationClient) *GRPCReplicationClient
  • func (rc *GRPCReplicationClient) Connect(ctx context.Context) error
  • func (rc *GRPCReplicationClient) Recv() ([]byte, error)
• type GRPCReplicationServer
  • func NewGRPCReplicationServer() *GRPCReplicationServer
  • func (rs *GRPCReplicationServer) GetWALStream(_ *pb.GetWALStreamRequest, stream pb.Replication_GetWALStreamServer) error
  • func (rs *GRPCReplicationServer) SendReplicationMessage(transactionGroup []byte)
• type Receiver
  • func NewReceiver(grpcClient GRPCClient, replayer Replayer) *Receiver
  • func (r *Receiver) Run(ctx context.Context) error
• type Replayer
• type ReplayerImpl
  • func NewReplayer(...) *ReplayerImpl
  • func (r *ReplayerImpl) Replay(transactionGroup []byte) error
• type Retryer
  • func NewRetryer(retryFunc func(ctx context.Context) error, interval time.Duration, ...) *Retryer
  • func (r *Retryer) Run(ctx context.Context) error
• type Sender
  • func NewSender(service Service) *Sender
  • func (s *Sender) Run(ctx context.Context)
  • func (s *Sender) Send(transactionGroup []byte)
• type Service

Variables

var ErrRetryable = errors.New("retryable replication error")

ErrRetryable is a custom error to retry the logic when returned.

Functions

func WTSetToCSM

func WTSetToCSM(wtSet *wal.WTSet) (io.ColumnSeriesMap, error)

WTSetToCSM converts wal.WTSet to ColumnSeriesMap.

Types

type GRPCClient

type GRPCClient interface {
	Connect(ctx context.Context) error
	Recv() ([]byte, error)
}

GRPCClient is an interface to abstract GRPCReplicationClient.

type GRPCReplicationClient

type GRPCReplicationClient struct {
	Client pb.ReplicationClient
	// contains filtered or unexported fields
}

func NewGRPCReplicationClient

func NewGRPCReplicationClient(client pb.ReplicationClient) *GRPCReplicationClient

func (*GRPCReplicationClient) Connect

func (rc *GRPCReplicationClient) Connect(ctx context.Context) error

func (*GRPCReplicationClient) Recv

func (rc *GRPCReplicationClient) Recv() ([]byte, error)

Recv blocks until it receives a response from gRPC stream connection.

type GRPCReplicationServer

type GRPCReplicationServer struct {
	pb.UnimplementedReplicationServer
	CertFile    string
	CertKeyFile string
	// Key: IPAddr (e.g. "192.125.18.1:25"), Value: channel for messages sent to each gRPC stream
	StreamChannels map[string]chan []byte
}

func NewGRPCReplicationServer

func NewGRPCReplicationServer() *GRPCReplicationServer

func (*GRPCReplicationServer) GetWALStream

func (rs *GRPCReplicationServer) GetWALStream(_ *pb.GetWALStreamRequest, stream pb.Replication_GetWALStreamServer,
) error

func (*GRPCReplicationServer) SendReplicationMessage

func (rs *GRPCReplicationServer) SendReplicationMessage(transactionGroup []byte)

type Receiver

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

func NewReceiver

func NewReceiver(grpcClient GRPCClient, replayer Replayer) *Receiver

func (*Receiver) Run

func (r *Receiver) Run(ctx context.Context) error

type Replayer

type Replayer interface {
	Replay(transactionGroup []byte) error
}

type ReplayerImpl

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

func NewReplayer

func NewReplayer(
	parseTGFunc func(tgSerialized []byte, rootPath string) (TGID int64, wtSets []wal.WTSet),
	writeFunc func(csm io.ColumnSeriesMap, isVariableLength bool) (err error),
	rootDir string,
) *ReplayerImpl

func (*ReplayerImpl) Replay

func (r *ReplayerImpl) Replay(transactionGroup []byte) error

type Retryer

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

func NewRetryer

func NewRetryer(retryFunc func(ctx context.Context) error, interval time.Duration, backoffCoeff int) *Retryer

func (*Retryer) Run

func (r *Retryer) Run(ctx context.Context) error

Run tries the Retryer until it succeeds, it returns unretriable error, or the context is canceled.

type Sender

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

func NewSender

func NewSender(service Service) *Sender

func (*Sender) Run

func (s *Sender) Run(ctx context.Context)

func (*Sender) Send

func (s *Sender) Send(transactionGroup []byte)

type Service

type Service interface {
	SendReplicationMessage(transactionGroup []byte)
}