buffer

README

etc data structure

Documentation

Index

• Variables
• func RunSegCloser(ctx context.Context, router *Router, allocator *domain.ChunkAllocator, ...)
• type Router
  • func NewRouter(ctx context.Context, segmentRepo *repo.SegmentRepo, ...) *Router
  • func (r *Router) Append(cmd *commands.Append) (err error)
  • func (r *Router) CloseSegment(addr buftypes.SegmentAddress) error
  • func (r *Router) CmdCh(addr buftypes.SegmentAddress) (chan commands.Command, error)
  • func (r *Router) DeleteTopic(topic types.Topic, size types.HashRangeSize)
  • func (r *Router) FinishSegments(topic types.Topic)
  • func (r *Router) Get(ctx context.Context, cmd *commands.Get) (err error)
  • func (r *Router) Poll(ctx context.Context, cmd *commands.Poll) error
  • func (r *Router) PutHashRangeSize(topic types.Topic, size types.HashRangeSize)
  • func (r *Router) PutSegment(addr buftypes.SegmentAddress, cmdCh chan commands.Command)
  • func (r *Router) SegmentFinished(topic types.Topic, hr types.HashRange, from uuid.Time)

Variables

var (
	ErrNotFound                 = errors.New("not found")
	ErrNotFoundInconsistentData = errors.New("not found, might be missing data")
	ErrInvalidSegmentAddress    = errors.New("invalid segment address")
)

Functions

func RunSegCloser

func RunSegCloser(ctx context.Context, router *Router, allocator *domain.ChunkAllocator, finishedSegmentsCh chan *buftypes.SegmentAddress)

Types

type Router

type Router struct {
	Ctx         context.Context
	Allocator   *domain.ChunkAllocator
	SegmentRepo *repo.SegmentRepo
	// contains filtered or unexported fields
}

func NewRouter

func NewRouter(ctx context.Context, segmentRepo *repo.SegmentRepo, allocator *domain.ChunkAllocator, finishedSegmentsCh chan *buftypes.SegmentAddress) *Router

func (*Router) Append

func (r *Router) Append(cmd *commands.Append) (err error)

func (*Router) CloseSegment

func (r *Router) CloseSegment(addr buftypes.SegmentAddress) error

func (*Router) CmdCh

func (r *Router) CmdCh(addr buftypes.SegmentAddress) (chan commands.Command, error)

func (*Router) DeleteTopic

func (r *Router) DeleteTopic(topic types.Topic, size types.HashRangeSize)

func (*Router) FinishSegments

func (r *Router) FinishSegments(topic types.Topic)

func (*Router) Get

func (r *Router) Get(ctx context.Context, cmd *commands.Get) (err error)

func (*Router) Poll

func (r *Router) Poll(ctx context.Context, cmd *commands.Poll) error

func (*Router) PutHashRangeSize

func (r *Router) PutHashRangeSize(topic types.Topic, size types.HashRangeSize)

func (*Router) PutSegment

func (r *Router) PutSegment(addr buftypes.SegmentAddress, cmdCh chan commands.Command)

func (*Router) SegmentFinished

func (r *Router) SegmentFinished(topic types.Topic, hr types.HashRange, from uuid.Time)