package topicreaderinternal

import "github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicreaderinternal"

Index ¶

• type PublicGetPartitionStartOffsetFunc
• type PublicGetPartitionStartOffsetRequest
• type PublicGetPartitionStartOffsetResponse
• func (r *PublicGetPartitionStartOffsetResponse) StartFrom(offset int64)
• type PublicReadBatchOption
• type PublicReaderOption
• func WithCredentials(cred credentials.Credentials) PublicReaderOption
• func WithTrace(tracer *trace.Topic) PublicReaderOption
• type ReadMessageBatchOptions
• type Reader
• func NewReader( client TopicClient, connector TopicSteamReaderConnect, consumer string, readSelectors []topicreadercommon.PublicReadSelector, opts ...PublicReaderOption, ) (Reader, error)
• func (r *Reader) Close(ctx context.Context) error
• func (r *Reader) Commit(ctx context.Context, offsets topicreadercommon.PublicCommitRangeGetter) (err error)
• func (r *Reader) CommitRanges(ctx context.Context, ranges []topicreadercommon.PublicCommitRange) error
• func (r *Reader) ID() int64
• func (r *Reader) PopBatchTx( ctx context.Context, tx tx.Transaction, opts ...PublicReadBatchOption, ) (*topicreadercommon.PublicBatch, error)
• func (r *Reader) ReadMessage(ctx context.Context) (*topicreadercommon.PublicMessage, error)
• func (r *Reader) ReadMessageBatch( ctx context.Context, opts ...PublicReadBatchOption, ) ( batch *topicreadercommon.PublicBatch, err error, )
• func (r *Reader) Tracer() *trace.Topic
• func (r *Reader) WaitInit(ctx context.Context) error
• type ReaderConfig
• func (cfg *ReaderConfig) Validate() []error
• type TopicClient
• type TopicSteamReaderConnect

Types ¶

type PublicGetPartitionStartOffsetFunc ¶

type PublicGetPartitionStartOffsetFunc func(
	ctx context.Context,
	req PublicGetPartitionStartOffsetRequest,
) (res PublicGetPartitionStartOffsetResponse, err error)

PublicGetPartitionStartOffsetFunc callback function for optional manage read progress store at own side

type PublicGetPartitionStartOffsetRequest ¶

type PublicGetPartitionStartOffsetRequest struct {
	Topic       string
	PartitionID int64

	// ExampleOnly
	PartitionSessionID int64
}

PublicGetPartitionStartOffsetRequest info about partition

type PublicGetPartitionStartOffsetResponse ¶

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

PublicGetPartitionStartOffsetResponse allow to set start offset for read messages for the partition

func (*PublicGetPartitionStartOffsetResponse) StartFrom ¶

func (r *PublicGetPartitionStartOffsetResponse) StartFrom(offset int64)

StartFrom set start offset for read the partition

type PublicReadBatchOption ¶

type PublicReadBatchOption interface {
	Apply(options ReadMessageBatchOptions) ReadMessageBatchOptions
}

PublicReadBatchOption для различных пожеланий к батчу вроде WithMaxMessages(int)

type PublicReaderOption ¶

type PublicReaderOption func(cfg *ReaderConfig)

func WithCredentials ¶

func WithCredentials(cred credentials.Credentials) PublicReaderOption

func WithTrace ¶

func WithTrace(tracer *trace.Topic) PublicReaderOption

type ReadMessageBatchOptions ¶

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

type Reader ¶

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

func NewReader ¶

func NewReader(
	client TopicClient,
	connector TopicSteamReaderConnect,
	consumer string,
	readSelectors []topicreadercommon.PublicReadSelector,
	opts ...PublicReaderOption,
) (Reader, error)

func (*Reader) Close ¶

func (r *Reader) Close(ctx context.Context) error

func (*Reader) Commit ¶

func (r *Reader) Commit(ctx context.Context, offsets topicreadercommon.PublicCommitRangeGetter) (err error)

func (*Reader) CommitRanges ¶

func (r *Reader) CommitRanges(ctx context.Context, ranges []topicreadercommon.PublicCommitRange) error

func (*Reader) ID ¶

func (r *Reader) ID() int64

func (*Reader) PopBatchTx ¶

func (r *Reader) PopBatchTx(
	ctx context.Context,
	tx tx.Transaction,
	opts ...PublicReadBatchOption,
) (*topicreadercommon.PublicBatch, error)

func (*Reader) ReadMessage ¶

func (r *Reader) ReadMessage(ctx context.Context) (*topicreadercommon.PublicMessage, error)

ReadMessage read exactly one message

func (*Reader) ReadMessageBatch ¶

func (r *Reader) ReadMessageBatch(
	ctx context.Context,
	opts ...PublicReadBatchOption,
) (
	batch *topicreadercommon.PublicBatch,
	err error,
)

ReadMessageBatch read batch of messages. Batch is collection of messages, which can be atomically committed

func (*Reader) Tracer ¶

func (r *Reader) Tracer() *trace.Topic

func (*Reader) WaitInit ¶

func (r *Reader) WaitInit(ctx context.Context) error

type ReaderConfig ¶

type ReaderConfig struct {
	config.Common

	RetrySettings      topic.RetrySettings
	DefaultBatchConfig ReadMessageBatchOptions
	// contains filtered or unexported fields
}

func (*ReaderConfig) Validate ¶

func (cfg *ReaderConfig) Validate() []error

type TopicClient ¶

type TopicClient interface {
	UpdateOffsetsInTransaction(ctx context.Context, req *rawtopic.UpdateOffsetsInTransactionRequest) error
}

TopicClient is part of rawtopic.Client

type TopicSteamReaderConnect ¶

type TopicSteamReaderConnect func(
	connectionCtx context.Context,
	readerID int64,
	tracer *trace.Topic,
) (topicreadercommon.RawTopicReaderStream, error)

TopicSteamReaderConnect connect to grpc stream when connectionCtx closed stream must stop work and return errors for all methods

Source Files ¶

batched_stream_reader_interface.go batcher.go public_callbacks.go reader.go stream_reader_impl.go stream_reconnector.go topic_client_interface.go