replication

Documentation

Index

• func NewRetryer(retryFunc func(ctx context.Context) error, interval time.Duration, ...) *retryer
• 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 NewGRPCReplicationService() *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 RetryableError
  • func (re RetryableError) Error() string
  • func (re RetryableError) Is(err error) bool
• type Sender
  • func NewSender(service Service) *Sender
  • func (s *Sender) Run(ctx context.Context)
  • func (s *Sender) Send(transactionGroup []byte)
• type Service

Functions

func NewRetryer

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

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 {
	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 NewGRPCReplicationService

func NewGRPCReplicationService() *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(TG_Serialized []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 RetryableError

type RetryableError string

RetryableError is a custom error to retry the logic when returned

func (RetryableError) Error

func (re RetryableError) Error() string

func (RetryableError) Is

func (re RetryableError) Is(err error) bool

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)
}