pbq

Documentation

Index

• Variables
• func DefaultOptions() *options
• type Manager
  • func NewManager(ctx context.Context, vertexName string, pipelineName string, vr int32, ...) (*Manager, error)
  • func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (ReadWriteCloser, error)
  • func (m *Manager) GetExistingPartitions(ctx context.Context) ([]partition.ID, error)
  • func (m *Manager) GetPBQ(partitionID partition.ID) ReadWriteCloser
  • func (m *Manager) ListPartitions() []*PBQ
  • func (m *Manager) NextWindowToBeMaterialized() window.AlignedWindower
  • func (m *Manager) Replay(ctx context.Context)
  • func (m *Manager) ShutDown(ctx context.Context)
• type PBQ
  • func (p *PBQ) Close() error
  • func (p *PBQ) CloseOfBook()
  • func (p *PBQ) GC() error
  • func (p *PBQ) ReadCh() <-chan *isb.ReadMessage
  • func (p *PBQ) Write(ctx context.Context, message *isb.ReadMessage) error
• type PBQOption
  • func WithChannelBufferSize(size int64) PBQOption
  • func WithReadBatchSize(size int64) PBQOption
  • func WithReadTimeout(seconds time.Duration) PBQOption
• type ReadWriteCloser
• type Reader
• type WriteCloser

Variables

var ErrCOB = errors.New("error while writing to pbq, pbq is closed")

Functions

func DefaultOptions

func DefaultOptions() *options

Types

type Manager

type Manager struct {

	// we need lock to access pbqMap, since deregister will be called inside pbq
	// and each pbq will be inside a go routine, and also entire PBQ could be managed
	// through a go routine (depends on the orchestrator)
	sync.RWMutex
	// contains filtered or unexported fields
}

Manager helps in managing the lifecycle of PBQ instances

func NewManager

func NewManager(ctx context.Context, vertexName string, pipelineName string, vr int32, storeProvider store.StoreProvider, opts ...PBQOption) (*Manager, error)

NewManager returns new instance of manager We don't intend this to be called by multiple routines.

func (*Manager) CreateNewPBQ

func (m *Manager) CreateNewPBQ(ctx context.Context, partitionID partition.ID) (ReadWriteCloser, error)

CreateNewPBQ creates new pbq for a partition

func (*Manager) GetExistingPartitions

func (m *Manager) GetExistingPartitions(ctx context.Context) ([]partition.ID, error)

GetExistingPartitions restores the state of the pbqManager. It reads from the PBQs store to get the persisted partitions and builds the PBQ Map.

func (*Manager) GetPBQ

func (m *Manager) GetPBQ(partitionID partition.ID) ReadWriteCloser

GetPBQ returns pbq for the given ID

func (*Manager) ListPartitions

func (m *Manager) ListPartitions() []*PBQ

ListPartitions returns all the pbq instances

func (*Manager) NextWindowToBeMaterialized

func (m *Manager) NextWindowToBeMaterialized() window.AlignedWindower

NextWindowToBeMaterialized returns the next keyed window that is yet to be materialized(GCed) will be used by the data forwarder to publish the idle watermark. While publishing idle watermark, we have to be conservative. PBQManager's view of next window to be materialized is conservative as it is on the reading side. We SHOULD NOT use NextWindowToBeMaterialized to write data to because it could fail (channel could have been closed but GC is yet to happen), this function should only be on readonly path.

func (*Manager) Replay

func (m *Manager) Replay(ctx context.Context)

Replay replays messages which are persisted in pbq store.

func (*Manager) ShutDown

func (m *Manager) ShutDown(ctx context.Context)

ShutDown for clean shut down, flushes pending messages to store and closes the store

type PBQ

type PBQ struct {
	PartitionID partition.ID
	// contains filtered or unexported fields
}

PBQ Buffer queue which is backed with a persisted store, each partition will have a PBQ associated with it

func (*PBQ) Close

func (p *PBQ) Close() error

Close is used by the writer to indicate close of context we should flush pending messages to store

func (*PBQ) CloseOfBook

func (p *PBQ) CloseOfBook()

CloseOfBook closes output channel

func (*PBQ) GC

func (p *PBQ) GC() error

GC cleans up the PBQ and also the store associated with it. GC is invoked after the Reader (ProcessAndForward) has finished forwarding the output to ISB.

func (*PBQ) ReadCh

func (p *PBQ) ReadCh() <-chan *isb.ReadMessage

ReadCh exposes read channel to read messages from PBQ close on read channel indicates COB

func (*PBQ) Write

func (p *PBQ) Write(ctx context.Context, message *isb.ReadMessage) error

Write writes message to pbq and persistent store

type PBQOption

type PBQOption func(options *options) error

func WithChannelBufferSize

func WithChannelBufferSize(size int64) PBQOption

WithChannelBufferSize sets buffer size option

func WithReadBatchSize

func WithReadBatchSize(size int64) PBQOption

WithReadBatchSize sets read batch size option

func WithReadTimeout

func WithReadTimeout(seconds time.Duration) PBQOption

WithReadTimeout sets read timeout option

type ReadWriteCloser

type ReadWriteCloser interface {
	Reader
	WriteCloser
}

ReadWriteCloser is an unified interface to PBQ read and write interfaces. Close is only for Writer.

type Reader

type Reader interface {
	// ReadCh exposes channel to read from PBQ
	ReadCh() <-chan *isb.ReadMessage
	// GC does garbage collection, it deletes all the persisted data from the store
	GC() error
}

Reader provides methods to read from PBQ.

type WriteCloser

type WriteCloser interface {
	// Write writes message to PBQ
	Write(ctx context.Context, msg *isb.ReadMessage) error
	// CloseOfBook (cob) closes PBQ, no writes will be accepted after cob
	CloseOfBook()
	// Close to handle context close on writer
	// Any pending data can be flushed to the persistent store at this point.
	Close() error
}

WriteCloser provides methods to write data to the PQB and close the PBQ. No data can be written to PBQ after cob.