producer

Documentation

Overview

Package producer is a generated GoMock package.

Index

• type Buffer
• type CloseType
• type FilterFunc
• type FinalizeReason
• type Message
• type MockMessage
  • func NewMockMessage(ctrl *gomock.Controller) *MockMessage
  • func (m *MockMessage) Bytes() []byte
  • func (m *MockMessage) EXPECT() *MockMessageMockRecorder
  • func (m *MockMessage) Finalize(arg0 FinalizeReason)
  • func (m *MockMessage) Shard() uint32
  • func (m *MockMessage) Size() int
• type MockMessageMockRecorder
  • func (mr *MockMessageMockRecorder) Bytes() *gomock.Call
  • func (mr *MockMessageMockRecorder) Finalize(arg0 interface{}) *gomock.Call
  • func (mr *MockMessageMockRecorder) Shard() *gomock.Call
  • func (mr *MockMessageMockRecorder) Size() *gomock.Call
• type MockProducer
  • func NewMockProducer(ctrl *gomock.Controller) *MockProducer
  • func (m *MockProducer) Close(arg0 CloseType)
  • func (m *MockProducer) EXPECT() *MockProducerMockRecorder
  • func (m *MockProducer) Init() error
  • func (m *MockProducer) NumShards() uint32
  • func (m *MockProducer) Produce(arg0 Message) error
  • func (m *MockProducer) RegisterFilter(arg0 services.ServiceID, arg1 FilterFunc)
  • func (m *MockProducer) UnregisterFilter(arg0 services.ServiceID)
• type MockProducerMockRecorder
  • func (mr *MockProducerMockRecorder) Close(arg0 interface{}) *gomock.Call
  • func (mr *MockProducerMockRecorder) Init() *gomock.Call
  • func (mr *MockProducerMockRecorder) NumShards() *gomock.Call
  • func (mr *MockProducerMockRecorder) Produce(arg0 interface{}) *gomock.Call
  • func (mr *MockProducerMockRecorder) RegisterFilter(arg0, arg1 interface{}) *gomock.Call
  • func (mr *MockProducerMockRecorder) UnregisterFilter(arg0 interface{}) *gomock.Call
• type OnFinalizeFn
• type Options
  • func NewOptions() Options
• type Producer
  • func NewProducer(opts Options) Producer
• type RefCountedMessage
  • func NewRefCountedMessage(m Message, fn OnFinalizeFn) *RefCountedMessage
  • func (rm *RefCountedMessage) Accept(fn FilterFunc) bool
  • func (rm *RefCountedMessage) DecReads()
  • func (rm *RefCountedMessage) DecRef()
  • func (rm *RefCountedMessage) Drop() bool
  • func (rm *RefCountedMessage) IncReads()
  • func (rm *RefCountedMessage) IncRef()
  • func (rm *RefCountedMessage) IsDroppedOrConsumed() bool
  • func (rm *RefCountedMessage) Size() uint64
• type Writer

Types

type Buffer

type Buffer interface {
	// Add adds message to the buffer and returns a reference counted message.
	Add(m Message) (*RefCountedMessage, error)

	// Init initializes the buffer.
	Init()

	// Close stops the buffer from accepting new requests immediately.
	// If the CloseType is WaitForConsumption, then it will block until all the messages have been consumed.
	// If the CloseType is DropEverything, then it will simply drop all the messages buffered and return.
	Close(ct CloseType)
}

Buffer buffers all the messages in the producer.

type CloseType

type CloseType int

CloseType decides how the producer should be closed.

const (
	// WaitForConsumption blocks the close call until all the messages have been consumed.
	WaitForConsumption CloseType = iota
	// DropEverything will close the producer and drop all the messages that have not been consumed.
	DropEverything
)

type FilterFunc

type FilterFunc func(m Message) bool

FilterFunc can filter message.

type FinalizeReason

type FinalizeReason int

FinalizeReason defines the reason why the message is being finalized by Producer.

const (
	// Consumed means the message has been fully consumed.
	Consumed FinalizeReason = iota

	// Dropped means the message has been dropped.
	Dropped
)

type Message

type Message interface {
	// Shard returns the shard of the message.
	Shard() uint32

	// Bytes returns the bytes of the message.
	Bytes() []byte

	// Size returns the size of the bytes of the message.
	Size() int

	// Finalize will be called by producer to indicate the end of its lifecycle.
	Finalize(FinalizeReason)
}

Message contains the data that will be produced by the producer. It should only be finalized by the producer.

type MockMessage

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

MockMessage is a mock of Message interface

func NewMockMessage

func NewMockMessage(ctrl *gomock.Controller) *MockMessage

NewMockMessage creates a new mock instance

func (*MockMessage) Bytes

func (m *MockMessage) Bytes() []byte

Bytes mocks base method

func (*MockMessage) EXPECT

func (m *MockMessage) EXPECT() *MockMessageMockRecorder

EXPECT returns an object that allows the caller to indicate expected use

func (*MockMessage) Finalize

func (m *MockMessage) Finalize(arg0 FinalizeReason)

Finalize mocks base method

func (*MockMessage) Shard

func (m *MockMessage) Shard() uint32

Shard mocks base method

func (*MockMessage) Size

func (m *MockMessage) Size() int

Size mocks base method

type MockMessageMockRecorder

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

MockMessageMockRecorder is the mock recorder for MockMessage

func (*MockMessageMockRecorder) Bytes

func (mr *MockMessageMockRecorder) Bytes() *gomock.Call

Bytes indicates an expected call of Bytes

func (*MockMessageMockRecorder) Finalize

func (mr *MockMessageMockRecorder) Finalize(arg0 interface{}) *gomock.Call

Finalize indicates an expected call of Finalize

func (*MockMessageMockRecorder) Shard

func (mr *MockMessageMockRecorder) Shard() *gomock.Call

Shard indicates an expected call of Shard

func (*MockMessageMockRecorder) Size

func (mr *MockMessageMockRecorder) Size() *gomock.Call

Size indicates an expected call of Size

type MockProducer

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

MockProducer is a mock of Producer interface

func NewMockProducer

func NewMockProducer(ctrl *gomock.Controller) *MockProducer

NewMockProducer creates a new mock instance

func (*MockProducer) Close

func (m *MockProducer) Close(arg0 CloseType)

Close mocks base method

func (*MockProducer) EXPECT

func (m *MockProducer) EXPECT() *MockProducerMockRecorder

EXPECT returns an object that allows the caller to indicate expected use

func (*MockProducer) Init

func (m *MockProducer) Init() error

Init mocks base method

func (*MockProducer) NumShards

func (m *MockProducer) NumShards() uint32

NumShards mocks base method

func (*MockProducer) Produce

func (m *MockProducer) Produce(arg0 Message) error

Produce mocks base method

func (*MockProducer) RegisterFilter

func (m *MockProducer) RegisterFilter(arg0 services.ServiceID, arg1 FilterFunc)

RegisterFilter mocks base method

func (*MockProducer) UnregisterFilter

func (m *MockProducer) UnregisterFilter(arg0 services.ServiceID)

UnregisterFilter mocks base method

type MockProducerMockRecorder

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

MockProducerMockRecorder is the mock recorder for MockProducer

func (*MockProducerMockRecorder) Close

func (mr *MockProducerMockRecorder) Close(arg0 interface{}) *gomock.Call

Close indicates an expected call of Close

func (*MockProducerMockRecorder) Init

func (mr *MockProducerMockRecorder) Init() *gomock.Call

Init indicates an expected call of Init

func (*MockProducerMockRecorder) NumShards

func (mr *MockProducerMockRecorder) NumShards() *gomock.Call

NumShards indicates an expected call of NumShards

func (*MockProducerMockRecorder) Produce

func (mr *MockProducerMockRecorder) Produce(arg0 interface{}) *gomock.Call

Produce indicates an expected call of Produce

func (*MockProducerMockRecorder) RegisterFilter

func (mr *MockProducerMockRecorder) RegisterFilter(arg0, arg1 interface{}) *gomock.Call

RegisterFilter indicates an expected call of RegisterFilter

func (*MockProducerMockRecorder) UnregisterFilter

func (mr *MockProducerMockRecorder) UnregisterFilter(arg0 interface{}) *gomock.Call

UnregisterFilter indicates an expected call of UnregisterFilter

type OnFinalizeFn

type OnFinalizeFn func(rm *RefCountedMessage)

OnFinalizeFn will be called when the message is being finalized.

type Options

type Options interface {
	// Buffer returns the buffer.
	Buffer() Buffer

	// SetBuffer sets the buffer.
	SetBuffer(value Buffer) Options

	// Writer returns the writer.
	Writer() Writer

	// SetWriter sets the writer.
	SetWriter(value Writer) Options
}

Options configs a producer.

func NewOptions

func NewOptions() Options

NewOptions creates a Options.

type Producer

type Producer interface {
	// Produce produces the message.
	Produce(m Message) error

	// RegisterFilter registers a filter to a consumer service.
	RegisterFilter(sid services.ServiceID, fn FilterFunc)

	// UnregisterFilter unregisters the filter of a consumer service.
	UnregisterFilter(sid services.ServiceID)

	// NumShards returns the total number of shards of the topic the producer is
	// producing to.
	NumShards() uint32

	// Init initializes a producer.
	Init() error

	// Close stops the producer from accepting new requests immediately.
	// If the CloseType is WaitForConsumption, then it will block until all the messages have been consumed.
	// If the CloseType is DropEverything, then it will simply drop all the messages buffered and return.
	Close(ct CloseType)
}

Producer produces message to a topic.

func NewProducer

func NewProducer(opts Options) Producer

NewProducer returns a new producer.

type RefCountedMessage

type RefCountedMessage struct {
	sync.RWMutex
	Message
	// contains filtered or unexported fields
}

RefCountedMessage is a reference counted message.

func NewRefCountedMessage

func NewRefCountedMessage(m Message, fn OnFinalizeFn) *RefCountedMessage

NewRefCountedMessage creates RefCountedMessage.

func (*RefCountedMessage) Accept

func (rm *RefCountedMessage) Accept(fn FilterFunc) bool

Accept returns true if the message can be accepted by the filter.

func (*RefCountedMessage) DecReads

func (rm *RefCountedMessage) DecReads()

DecReads decrements the reads count.

func (*RefCountedMessage) DecRef

func (rm *RefCountedMessage) DecRef()

DecRef decrements the ref count. If the reference count became zero after the call, the message will be finalized as consumed.

func (*RefCountedMessage) Drop

func (rm *RefCountedMessage) Drop() bool

Drop drops the message without waiting for it to be consumed.

func (*RefCountedMessage) IncReads

func (rm *RefCountedMessage) IncReads()

IncReads increments the reads count.

func (*RefCountedMessage) IncRef

func (rm *RefCountedMessage) IncRef()

IncRef increments the ref count.

func (*RefCountedMessage) IsDroppedOrConsumed

func (rm *RefCountedMessage) IsDroppedOrConsumed() bool

IsDroppedOrConsumed returns true if the message has been dropped or consumed.

func (*RefCountedMessage) Size

func (rm *RefCountedMessage) Size() uint64

Size returns the size of the message.

type Writer

type Writer interface {
	// Write writes a reference counted message out.
	Write(rm *RefCountedMessage) error

	// RegisterFilter registers a filter to a consumer service.
	RegisterFilter(sid services.ServiceID, fn FilterFunc)

	// UnregisterFilter unregisters the filter of a consumer service.
	UnregisterFilter(sid services.ServiceID)

	// NumShards returns the total number of shards of the topic the writer is
	// writing to.
	NumShards() uint32

	// Init initializes a writer.
	Init() error

	// Close closes the writer.
	Close()
}

Writer writes all the messages out to the consumer services.