server

Documentation

Index

• Constants
• type OnDisk
  • func NewOnDisk(logger *log.Logger, dirname, category, instanceName string, ...) (*OnDisk, error)
  • func (s *OnDisk) Ack(ctx context.Context, chunk string, size uint64) error
  • func (s *OnDisk) AckDirect(chunk string) error
  • func (s *OnDisk) ListChunks() ([]protocol.Chunk, error)
  • func (s *OnDisk) Read(chunk string, off uint64, maxSize uint64, w io.Writer) error
  • func (s *OnDisk) ReplicationAck(ctx context.Context, chunk, instance string, size uint64) error
  • func (s *OnDisk) SetReplicationDisabled(v bool)
  • func (s *OnDisk) Wait(ctx context.Context, chunkName string, off uint64, minSyncReplicas uint) error
  • func (s *OnDisk) Write(ctx context.Context, msgs []byte) (chunkName string, off int64, err error)
  • func (s *OnDisk) WriteDirect(chunk string, contents []byte) error
• type OnDiskCreator
  • func NewOnDiskCreator(logger *log.Logger, dirName string, instanceName string, ...) *OnDiskCreator
  • func (c *OnDiskCreator) AckDirect(ctx context.Context, category string, chunk string) error
  • func (c *OnDiskCreator) Get(category string) (*OnDisk, error)
  • func (c *OnDiskCreator) SetReplicationDisabled(category string, v bool) error
  • func (c *OnDiskCreator) Stat(category string, fileName string) (size int64, exists bool, deleted bool, err error)
  • func (c *OnDiskCreator) Write(ctx context.Context, category string, msgs []byte) (chunkName string, off int64, err error)
  • func (c *OnDiskCreator) WriteDirect(category string, fileName string, contents []byte) error
• type StorageHooks

Constants

const AcknowledgedSuffix = ".acknowledged"

Types

type OnDisk

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

OnDisk stores all the data on disk.

func NewOnDisk

func NewOnDisk(logger *log.Logger, dirname, category, instanceName string, maxChunkSize uint64, maxLineSize uint64, rotateChunkInterval time.Duration, repl StorageHooks) (*OnDisk, error)

NewOnDisk creates a server that stores all it's data on disk.

func (*OnDisk) Ack

func (s *OnDisk) Ack(ctx context.Context, chunk string, size uint64) error

Ack marks the current chunk as done and deletes it's contents.

func (*OnDisk) AckDirect

func (s *OnDisk) AckDirect(chunk string) error

AckDirect is a method that is called from replication to replay acknowledge requests on the replica side.

func (*OnDisk) ListChunks

func (s *OnDisk) ListChunks() ([]protocol.Chunk, error)

ListChunks returns the list of current chunks.

func (*OnDisk) Read

func (s *OnDisk) Read(chunk string, off uint64, maxSize uint64, w io.Writer) error

Read copies the data from the in-memory store and writes the data read to the provided Writer, starting with the offset provided.

func (*OnDisk) ReplicationAck

func (s *OnDisk) ReplicationAck(ctx context.Context, chunk, instance string, size uint64) error

ReplicationAck lets writers that are waiting for min_sync_replicas to know that their writes were successful.

func (*OnDisk) SetReplicationDisabled

func (s *OnDisk) SetReplicationDisabled(v bool)

func (*OnDisk) Wait

func (s *OnDisk) Wait(ctx context.Context, chunkName string, off uint64, minSyncReplicas uint) error

Wait waits until the minSyncReplicas report back that they successfully downloaded the respective chunk from us.

func (*OnDisk) Write

func (s *OnDisk) Write(ctx context.Context, msgs []byte) (chunkName string, off int64, err error)

Write accepts the messages from the clients and stores them.

func (*OnDisk) WriteDirect

func (s *OnDisk) WriteDirect(chunk string, contents []byte) error

WriteDirect writes directly to the chunk files to avoid circular dependency with replication. THE METHOD IS NOT THREAD-SAFE.

type OnDiskCreator

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

func NewOnDiskCreator

func NewOnDiskCreator(logger *log.Logger, dirName string, instanceName string, replStorage *replication.Storage, maxChunkSize uint64, rotateChunkInterval time.Duration) *OnDiskCreator

func (*OnDiskCreator) AckDirect

func (c *OnDiskCreator) AckDirect(ctx context.Context, category string, chunk string) error

func (*OnDiskCreator) Get

func (c *OnDiskCreator) Get(category string) (*OnDisk, error)

func (*OnDiskCreator) SetReplicationDisabled

func (c *OnDiskCreator) SetReplicationDisabled(category string, v bool) error

func (*OnDiskCreator) Stat

func (c *OnDiskCreator) Stat(category string, fileName string) (size int64, exists bool, deleted bool, err error)

Stat returns information about the chunk: whether or not it exists and it's size. If file does not exist no error is returned.

func (*OnDiskCreator) Write

func (c *OnDiskCreator) Write(ctx context.Context, category string, msgs []byte) (chunkName string, off int64, err error)

func (*OnDiskCreator) WriteDirect

func (c *OnDiskCreator) WriteDirect(category string, fileName string, contents []byte) error

type StorageHooks

type StorageHooks interface {
	AfterCreatingChunk(ctx context.Context, category string, fileName string)
	AfterAcknowledgeChunk(ctx context.Context, category string, fileName string)
}