cmdhandler

Documentation

Index

• Constants
• Variables
• type Handler
  • func NewHandler(ctx context.Context, service *SegmentService) *Handler
  • func (h *Handler) Close()
  • func (h *Handler) Handle(cmd commands.Command)
  • func (h *Handler) Run()
• type SegmentService
  • func NewSegmentService(topic types.Topic, hashRange types.HashRange, props SegmentServiceProps) *SegmentService
  • func (s *SegmentService) Append(cmd *commands.Append) error
  • func (s *SegmentService) Finish()
  • func (s *SegmentService) Get(key types.Key) (msgId uuid.UUID, data []byte, ok bool, err error)
  • func (s *SegmentService) Poll(addr buftypes.SegmentAddress, from time.Time, fromMsgId uuid.UUID) (domain.PollResult, error)
  • func (s *SegmentService) String() string
• type SegmentServiceProps

Constants

const (
	MaxBufSize          = 1<<20*10 - 1024 // Just under 10MB
	MaxBufferedCommands = 40_000
)

Variables

var (
	ErrSegmentClosed = errors.New("segment closed")
	ErrCancelled     = errors.New("command cancelled")
)

var (
	ErrInvalidSegmentAddress = errors.New("invalid segment address")
	ErrBufferSizeExceeded    = errors.New("buffer size exceeded")
	ErrSegmentFinished       = errors.New("segment finished")
)

Types

type Handler

type Handler struct {
	Ctx     context.Context
	CmdCh   chan commands.Command
	Service *SegmentService
	Conn    *bq.BigQueryConn
	Next    *SegmentService

	AppendCh         chan struct{}
	AppendFinishedCh chan error
	// contains filtered or unexported fields
}

func NewHandler

func NewHandler(ctx context.Context, service *SegmentService) *Handler

func (*Handler) Close

func (h *Handler) Close()

func (*Handler) Handle

func (h *Handler) Handle(cmd commands.Command)

func (*Handler) Run

func (h *Handler) Run()

type SegmentService

type SegmentService struct {
	SegmentServiceProps
	Segment            domain.Segment
	CmdCh              chan commands.Command
	Initialized        bool
	UncommittedAppends chan *commands.Append
	Writer             *bq.BatchWriter
}

func NewSegmentService

func NewSegmentService(topic types.Topic, hashRange types.HashRange, props SegmentServiceProps) *SegmentService

func (*SegmentService) Append

func (s *SegmentService) Append(cmd *commands.Append) error

func (*SegmentService) Finish

func (s *SegmentService) Finish()

func (*SegmentService) Get

func (s *SegmentService) Get(key types.Key) (msgId uuid.UUID, data []byte, ok bool, err error)

func (*SegmentService) Poll

func (s *SegmentService) Poll(addr buftypes.SegmentAddress, from time.Time, fromMsgId uuid.UUID) (domain.PollResult, error)

TODO We might want the poll result to be a channel and get notified if append is called. TODO If we are at the end of the segment, segment is not finished, and we have no messages, we should wait for a message to be appended. TODO We need to be notified if: client ctx is Done, segment is finished, segment is closed, and message is appended.

func (*SegmentService) String

func (s *SegmentService) String() string

type SegmentServiceProps

type SegmentServiceProps struct {
	Ctx          context.Context
	Router       *buffer.Router
	Repo         *repo.SegmentRepo
	Conn         *bq.BigQueryConn
	FlushEnabled bool
}