sync

Documentation

Index

• Constants
• Variables
• type HandleMessage
  • func (m HandleMessage) LogFields(fields ...zap.Field) []zap.Field
• type Limit
  • func NewLimit(peerStep, totalStep []int, excludeIds []string, excludedLimit int) *Limit
  • func (l *Limit) Release(id string)
  • func (l *Limit) Stats(id string) string
  • func (l *Limit) Take(id string) bool
• type RequestManager
  • func NewRequestManager(handler syncdeps.SyncHandler, metric syncdeps.QueueSizeUpdater, ...) RequestManager
• type RequestPool
  • func NewRequestPool(closePeriod, gcPeriod time.Duration) RequestPool
• type StreamResponse
• type SyncService
  • func NewSyncService() SyncService

Constants

const CName = "common.commonspace.sync"

Variables

var ErrUnexpectedMessage = errors.New("unexpected message")

Types

type HandleMessage

type HandleMessage struct {
	Id                uint64
	ReceiveTime       time.Time
	StartHandlingTime time.Time
	Deadline          time.Time
	SenderId          string
	Message           *spacesyncproto.ObjectSyncMessage
	PeerCtx           context.Context
}

func (HandleMessage) LogFields

func (m HandleMessage) LogFields(fields ...zap.Field) []zap.Field

type Limit

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

func NewLimit

func NewLimit(peerStep, totalStep []int, excludeIds []string, excludedLimit int) *Limit

func (*Limit) Release

func (l *Limit) Release(id string)

func (*Limit) Stats

func (l *Limit) Stats(id string) string

func (*Limit) Take

func (l *Limit) Take(id string) bool

type RequestManager

type RequestManager interface {
	QueueRequest(rq syncdeps.Request) error
	SendRequest(ctx context.Context, rq syncdeps.Request, collector syncdeps.ResponseCollector) error
	HandleStreamRequest(ctx context.Context, rq syncdeps.Request, stream drpc.Stream) error
	HandleDeprecatedObjectSync(ctx context.Context, req *spacesyncproto.ObjectSyncMessage) (resp *spacesyncproto.ObjectSyncMessage, err error)
	Close()
}

func NewRequestManager

func NewRequestManager(handler syncdeps.SyncHandler, metric syncdeps.QueueSizeUpdater, responsibleNodeIds []string) RequestManager

type RequestPool

type RequestPool interface {
	TryTake(peerId, objectId string) bool
	Release(peerId, objectId string)
	QueueRequestAction(peerId, objectId string, action func(ctx context.Context), remove func()) (err error)
	Close()
}

func NewRequestPool

func NewRequestPool(closePeriod, gcPeriod time.Duration) RequestPool

type StreamResponse

type StreamResponse struct {
	Stream     drpc.Stream
	Connection drpc.Conn
}

type SyncService

type SyncService interface {
	app.Component
	BroadcastMessage(ctx context.Context, msg drpc.Message) error
	HandleDeprecatedObjectSync(ctx context.Context, req *spacesyncproto.ObjectSyncMessage) (resp *spacesyncproto.ObjectSyncMessage, err error)
	HandleStreamRequest(ctx context.Context, req syncdeps.Request, stream drpc.Stream) error
	SendRequest(ctx context.Context, rq syncdeps.Request, collector syncdeps.ResponseCollector) error
	QueueRequest(ctx context.Context, rq syncdeps.Request) error
	CloseReceiveQueue(id string) error
}

func NewSyncService

func NewSyncService() SyncService