package topicreaderinternal

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

Index ¶

• Variables
• type PublicCommitMode
• 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 RawTopicReaderStream
• type ReadMessageBatchOptions
• type Reader
• func NewReader( 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) 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 TopicSteamReaderConnect

Variables ¶

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

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

Types ¶

type PublicCommitMode ¶

type PublicCommitMode int

const (
	CommitModeAsync PublicCommitMode = iota // default
	CommitModeNone
	CommitModeSync
)

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 RawTopicReaderStream ¶

type RawTopicReaderStream interface {
	Recv() (rawtopicreader.ServerMessage, error)
	Send(msg rawtopicreader.ClientMessage) error
	CloseSend() error
}

type ReadMessageBatchOptions ¶

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

type Reader ¶

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

func NewReader ¶

func NewReader(
	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) 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 TopicSteamReaderConnect ¶

type TopicSteamReaderConnect func(connectionCtx context.Context) (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 committer.go grpc_synced_stream.go public_callbacks.go reader.go stream_reader_impl.go stream_reconnector.go