topicreadercommon

Sep 5, 2024

Changes in this version

+ var ErrCommitDisabled = xerrors.Wrap(errors.New("ydb: commits disabled"))

+ var ErrWrongCommitOrderInSyncMode = xerrors.Wrap(errors.New("ydb: wrong commit order in sync mode"))

+ var PublicErrCommitSessionToExpiredSession = xerrors.Wrap(errors.New("ydb: commit to expired session"))

type CommitRanges

+ func (r *CommitRanges) ToRawMessage() *rawtopicreader.CommitOffsetRequest

+ type Committer struct

+ BufferCountTrigger int

+ BufferTimeLagTrigger time.Duration

+ func NewCommitterStopped(tracer *trace.Topic, lifeContext context.Context, mode PublicCommitMode, ...) *Committer

+ func (c *Committer) Close(ctx context.Context, err error) error

+ func (c *Committer) Commit(ctx context.Context, commitRange CommitRange) error

+ func (c *Committer) OnCommitNotify(session *PartitionSession, offset rawtopiccommon.Offset)

+ func (c *Committer) Start()

+ type PublicCommitMode int

+ const CommitModeAsync

+ const CommitModeNone

+ const CommitModeSync

+ func (m PublicCommitMode) CommitsEnabled() bool

+ type RawTopicReaderStream interface

+ CloseSend func() error

+ Recv func() (rawtopicreader.ServerMessage, error)

+ Send func(msg rawtopicreader.ClientMessage) error

+ type SendMessageToServerFunc func(msg rawtopicreader.ClientMessage) error

+ type SyncedStream struct

+ func NewSyncedStream(stream RawTopicReaderStream) *SyncedStream

+ func (s *SyncedStream) CloseSend() error

+ func (s *SyncedStream) Recv() (rawtopicreader.ServerMessage, error)

+ func (s *SyncedStream) Send(msg rawtopicreader.ClientMessage) error

Aug 9, 2024

Changes in this version

type PartitionSession

+ func (s *PartitionSession) SetCommittedOffsetForward(v rawtopiccommon.Offset)

type PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) PartitionSession(session *PartitionSession) *PublicMessageBuilder

Jul 26, 2024

Changes in this version

+ const DefaultBufferSize

+ var ErrPublicUnexpectedCodec = xerrors.Wrap(errors.New("ydb: unexpected codec"))

+ func BatchIsEmpty(b *PublicBatch) bool

+ func CreateInitMessage(consumer string, selectors []*PublicReadSelector) *rawtopicreader.InitRequest

+ func MessageGetBufferBytesAccount(m *PublicMessage) int

+ func MessageSetNilDataForTest(m *PublicMessage)

+ func NextReaderID() int64

+ type CallbackWithMessageContentFunc func(data []byte) error

+ type CommitRange struct

+ CommitOffsetEnd rawtopicreader.Offset

+ CommitOffsetStart rawtopicreader.Offset

+ PartitionSession *PartitionSession

+ func GetCommitRange(item PublicCommitRangeGetter) CommitRange

+ type CommitRanges struct

+ Ranges []CommitRange

+ func NewCommitRangesFromPublicCommits(ranges []PublicCommitRange) CommitRanges

+ func NewCommitRangesWithCapacity(capacity int) CommitRanges

+ func (r *CommitRanges) Append(ranges ...PublicCommitRangeGetter)

+ func (r *CommitRanges) AppendCommitRange(cr CommitRange)

+ func (r *CommitRanges) AppendCommitRanges(ranges []CommitRange)

+ func (r *CommitRanges) AppendMessages(messages ...PublicMessage)

+ func (r *CommitRanges) GetCommitsInfo() []trace.TopicReaderStreamCommitInfo

+ func (r *CommitRanges) Len() int

+ func (r *CommitRanges) Optimize()

+ func (r *CommitRanges) Reset()

+ func (r *CommitRanges) ToPartitionsOffsets() []rawtopicreader.PartitionCommitOffset

+ type DecoderMap struct

+ func NewDecoderMap() DecoderMap

+ func (m *DecoderMap) AddDecoder(codec rawtopiccommon.Codec, createFunc PublicCreateDecoderFunc)

+ func (m *DecoderMap) Decode(codec rawtopiccommon.Codec, input io.Reader) (io.Reader, error)

+ type PartitionSession struct

+ ClientPartitionSessionID int64

+ PartitionID int64

+ ReaderID int64

+ StreamPartitionSessionID rawtopicreader.PartitionSessionID

+ Topic string

+ func BatchGetPartitionSession(item *PublicBatch) *PartitionSession

+ func NewPartitionSession(partitionContext context.Context, topic string, partitionID int64, ...) *PartitionSession

+ func (s *PartitionSession) Close()

+ func (s *PartitionSession) CommittedOffset() rawtopicreader.Offset

+ func (s *PartitionSession) Context() context.Context

+ func (s *PartitionSession) LastReceivedMessageOffset() rawtopicreader.Offset

+ func (s *PartitionSession) SetCommittedOffset(v rawtopicreader.Offset)

+ func (s *PartitionSession) SetContext(ctx context.Context)

+ func (s *PartitionSession) SetLastReceivedMessageOffset(v rawtopicreader.Offset)

+ func (s *PartitionSession) ToPublic() PublicPartitionSession

+ type PartitionSessionStorage struct

+ func (c *PartitionSessionStorage) Add(session *PartitionSession) error

+ func (c *PartitionSessionStorage) Get(id rawtopicreader.PartitionSessionID) (*PartitionSession, error)

+ func (c *PartitionSessionStorage) GetAll() []*PartitionSession

+ func (c *PartitionSessionStorage) Remove(id rawtopicreader.PartitionSessionID) (*PartitionSession, error)

+ type Pool interface

+ Get func() interface{}

+ Put func(x interface{})

+ type PublicBatch struct

+ Messages []*PublicMessage

+ func BatchAppend(original, appended *PublicBatch) (*PublicBatch, error)

+ func BatchCutMessages(b *PublicBatch, count int) (head, rest *PublicBatch)

+ func BatchSetCommitRangeForTest(b *PublicBatch, commitRange CommitRange) *PublicBatch

+ func NewBatch(session *PartitionSession, messages []*PublicMessage) (*PublicBatch, error)

+ func NewBatchFromStream(decoders DecoderMap, session *PartitionSession, sb rawtopicreader.Batch) (*PublicBatch, error)

+ func ReadRawBatchesToPublicBatches(msg *rawtopicreader.ReadResponse, sessions *PartitionSessionStorage, ...) ([]*PublicBatch, error)

+ func (m *PublicBatch) Context() context.Context

+ func (m *PublicBatch) PartitionID() int64

+ func (m *PublicBatch) Topic() string

+ type PublicCommitRange struct

+ type PublicCommitRangeGetter interface

+ type PublicCreateDecoderFunc func(input io.Reader) (io.Reader, error)

+ type PublicMessage struct

+ CreatedAt time.Time

+ MessageGroupID string

+ Metadata map[string][]byte

+ Offset int64

+ ProducerID string

+ SeqNo int64

+ UncompressedSize int

+ WriteSessionMetadata map[string]string

+ WrittenAt time.Time

+ func MessageWithSetCommitRangeForTest(m *PublicMessage, commitRange CommitRange) *PublicMessage

+ func (m *PublicMessage) Context() context.Context

+ func (m *PublicMessage) PartitionID() int64

+ func (m *PublicMessage) Read(p []byte) (n int, err error)

+ func (m *PublicMessage) Topic() string

+ func (m *PublicMessage) UnmarshalTo(dst PublicMessageContentUnmarshaler) error

+ type PublicMessageBuilder struct

+ func NewPublicMessageBuilder() *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Build() *PublicMessage

+ func (pmb *PublicMessageBuilder) CommitRange(cr CommitRange) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Context(ctx context.Context)

+ func (pmb *PublicMessageBuilder) CreatedAt(createdAt time.Time) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) DataAndUncompressedSize(data []byte) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) MessageGroupID(messageGroupID string) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Metadata(metadata map[string][]byte) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Offset(offset int64) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) PartitionID(partitionID int64)

+ func (pmb *PublicMessageBuilder) ProducerID(producerID string) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) RawDataLen(val int) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Seqno(seqNo int64) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) Topic(topic string)

+ func (pmb *PublicMessageBuilder) UncompressedSize(uncompressedSize int) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) WriteSessionMetadata(writeSessionMetadata map[string]string) *PublicMessageBuilder

+ func (pmb *PublicMessageBuilder) WrittenAt(writtenAt time.Time) *PublicMessageBuilder

+ type PublicMessageContentUnmarshaler interface

+ UnmarshalYDBTopicMessage func(data []byte) error

+ type PublicPartitionSession struct

+ PartitionID int64

+ PartitionSessionID int64

+ TopicPath string

+ type PublicReadSelector struct

+ MaxTimeLag time.Duration

+ Partitions []int64

+ Path string

+ ReadFrom time.Time

+ func (s PublicReadSelector) Clone() *PublicReadSelector