candidatebuffer

Documentation

Index

• type CandidateBuffer
  • func NewCandidateBuffer(onCandidates func([]types.RetrievalCandidate), clock clock.Clock) *CandidateBuffer
  • func NewCandidateBufferWithSync(onCandidates func([]types.RetrievalCandidate), clock clock.Clock, ...) *CandidateBuffer
  • func (c *CandidateBuffer) BufferStream(ctx context.Context, incoming <-chan types.FindCandidatesResult, ...) error

Types

type CandidateBuffer

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

func NewCandidateBuffer

func NewCandidateBuffer(onCandidates func([]types.RetrievalCandidate), clock clock.Clock) *CandidateBuffer

func NewCandidateBufferWithSync

func NewCandidateBufferWithSync(onCandidates func([]types.RetrievalCandidate), clock clock.Clock, afterEach chan<- struct{}) *CandidateBuffer

func (*CandidateBuffer) BufferStream

func (c *CandidateBuffer) BufferStream(ctx context.Context, incoming <-chan types.FindCandidatesResult, bufferingTime time.Duration) error

BufferStream consumes a stream of individual candidate results. When a new result comes in, a collection is started, and further results are added to the collection until the specified bufferingTime has passed, at which point the collection is passed to the callback setup when the Buffer was setup. The timer is reset and the collection emptied until another result comes in. This has the effect of grouping results that occur in the same general time frame.