replication

Documentation

Index

• Constants
• type CategoryDownloader
  • func (c *CategoryDownloader) Loop(ctx context.Context)
• type Chunk
• type Client
  • func NewClient(logger *log.Logger, dirname string, wr DirectWriter, peers []Peer, ...) *Client
  • func (c *Client) Loop(ctx context.Context)
• type DirectWriter
• type Message
• type MessageType
• type Peer
• type Storage
  • func NewStorage(logger *log.Logger, currentInstance string) *Storage
  • func (s *Storage) AfterAcknowledgeChunk(ctx context.Context, category string, fileName string)
  • func (s *Storage) AfterCreatingChunk(ctx context.Context, category string, fileName string)
  • func (s *Storage) RegisterReplica(instanceName string, ch chan Message)

Constants

const BatchSize = 4 * 1024 * 1024 // 4 MiB

BatchSize is the max number of bytes that can be downloaded in a single request in replication.

Types

type CategoryDownloader

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

func (*CategoryDownloader) Loop

func (c *CategoryDownloader) Loop(ctx context.Context)

type Chunk

type Chunk struct {
	Owner    string
	Category string
	FileName string
}

type Client

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

Client describes the client-side state of replication and continiously downloads new chunks from other servers.

func NewClient

func NewClient(logger *log.Logger, dirname string, wr DirectWriter, peers []Peer, instanceName string) *Client

NewClient initialises the replication client.

func (*Client) Loop

func (c *Client) Loop(ctx context.Context)

type DirectWriter

type DirectWriter interface {
	Stat(category string, fileName string) (size int64, exists bool, deleted bool, err error)
	WriteDirect(category string, fileName string, contents []byte) error
	AckDirect(ctx context.Context, category string, chunk string) error

	SetReplicationDisabled(category string, v bool) error
	Write(ctx context.Context, category string, msgs []byte) (chunkName string, off int64, err error)
}

DirectWriter writes to underlying storage directly for replication purposes.

type Message

type Message struct {
	Type  MessageType
	Chunk Chunk
}

type MessageType

type MessageType string

const (
	ChunkCreated      MessageType = "chunk_created"
	ChunkAcknowledged MessageType = "chunk_ack"
)

type Peer

type Peer struct {
	InstanceName string
	ListenAddr   string
}

type Storage

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

Storage provides hooks for the ondisk storage that will be called to ensure that chunks are replicated.

func NewStorage

func NewStorage(logger *log.Logger, currentInstance string) *Storage

func (*Storage) AfterAcknowledgeChunk

func (s *Storage) AfterAcknowledgeChunk(ctx context.Context, category string, fileName string)

func (*Storage) AfterCreatingChunk

func (s *Storage) AfterCreatingChunk(ctx context.Context, category string, fileName string)

func (*Storage) RegisterReplica

func (s *Storage) RegisterReplica(instanceName string, ch chan Message)