rawtopicreader

Documentation

Index

• Variables
• type Batch
• type ClientMessage
• type CommitOffsetRequest
• type CommitOffsetResponse
• type GrpcStream
• type InitRequest
  • func (r *InitRequest) GetConsumer() string
  • func (r *InitRequest) GetTopics() []string
• type InitResponse
• type MessageData
• type Offset
  • func NewOffset(v int64) Offset
  • func (offset *Offset) FromInt64(v int64)
  • func (offset Offset) ToInt64() int64
• type OffsetRange
  • func (r *OffsetRange) FromProto(p *Ydb_Topic.OffsetsRange) error
  • func (r *OffsetRange) ToProto() *Ydb_Topic.OffsetsRange
• type OptionalOffset
  • func (offset *OptionalOffset) FromInt64(v int64)
  • func (offset *OptionalOffset) FromInt64Pointer(v *int64)
  • func (offset OptionalOffset) ToInt64() int64
  • func (offset OptionalOffset) ToInt64Pointer() *int64
• type PartitionCommitOffset
• type PartitionCommittedOffset
• type PartitionData
• type PartitionSession
• type PartitionSessionID
  • func (id *PartitionSessionID) FromInt64(v int64)
  • func (id PartitionSessionID) ToInt64() int64
• type PartitionSessionStatusRequest
• type PartitionSessionStatusResponse
• type ReadRequest
• type ReadResponse
  • func (r *ReadResponse) GetBytesSize() int
  • func (r *ReadResponse) GetPartitionBatchMessagesCounts() (partitionDataCount, batchCount, messagesCount int)
• type ServerMessage
• type StartPartitionSessionRequest
• type StartPartitionSessionResponse
• type StopPartitionSessionRequest
• type StopPartitionSessionResponse
• type StreamReader
  • func (s StreamReader) CloseSend() error
  • func (s StreamReader) Recv() (ServerMessage, error)
  • func (s StreamReader) Send(msg ClientMessage) (err error)
• type TopicReadSettings
• type UpdateTokenRequest
• type UpdateTokenResponse

Variables

var ErrUnexpectedMessageType = errors.New("unexpected message type")

Types

type Batch

type Batch struct {
	Codec rawtopiccommon.Codec

	ProducerID       string
	WriteSessionMeta map[string]string // nil if session meta is empty
	WrittenAt        time.Time

	MessageData []MessageData
}

type ClientMessage

type ClientMessage interface {
	// contains filtered or unexported methods
}

type CommitOffsetRequest

type CommitOffsetRequest struct {
	CommitOffsets []PartitionCommitOffset
	// contains filtered or unexported fields
}

type CommitOffsetResponse

type CommitOffsetResponse struct {
	rawtopiccommon.ServerMessageMetadata

	PartitionsCommittedOffsets []PartitionCommittedOffset
	// contains filtered or unexported fields
}

type GrpcStream

type GrpcStream interface {
	Send(messageNew *Ydb_Topic.StreamReadMessage_FromClient) error
	Recv() (*Ydb_Topic.StreamReadMessage_FromServer, error)
	CloseSend() error
}

type InitRequest

type InitRequest struct {
	TopicsReadSettings []TopicReadSettings

	Consumer string
	// contains filtered or unexported fields
}

func (*InitRequest) GetConsumer

func (r *InitRequest) GetConsumer() string

GetConsumer for implement trace.TopicReadStreamInitRequestInfo

func (*InitRequest) GetTopics

func (r *InitRequest) GetTopics() []string

GetTopics for implement trace.TopicReadStreamInitRequestInfo

type InitResponse

type InitResponse struct {
	rawtopiccommon.ServerMessageMetadata

	SessionID string
	// contains filtered or unexported fields
}

type MessageData

type MessageData struct {
	Offset           Offset
	SeqNo            int64
	CreatedAt        time.Time
	Data             []byte
	UncompressedSize int64
	MessageGroupID   string
	MetadataItems    []rawtopiccommon.MetadataItem
}

type Offset

type Offset int64

func NewOffset

func NewOffset(v int64) Offset

func (*Offset) FromInt64

func (offset *Offset) FromInt64(v int64)

func (Offset) ToInt64

func (offset Offset) ToInt64() int64

type OffsetRange

type OffsetRange struct {
	Start Offset
	End   Offset
}

func (*OffsetRange) FromProto

func (r *OffsetRange) FromProto(p *Ydb_Topic.OffsetsRange) error

func (*OffsetRange) ToProto

func (r *OffsetRange) ToProto() *Ydb_Topic.OffsetsRange

type OptionalOffset

type OptionalOffset struct {
	Offset   Offset
	HasValue bool
}

func (*OptionalOffset) FromInt64

func (offset *OptionalOffset) FromInt64(v int64)

func (*OptionalOffset) FromInt64Pointer

func (offset *OptionalOffset) FromInt64Pointer(v *int64)

func (OptionalOffset) ToInt64

func (offset OptionalOffset) ToInt64() int64

func (OptionalOffset) ToInt64Pointer

func (offset OptionalOffset) ToInt64Pointer() *int64

type PartitionCommitOffset

type PartitionCommitOffset struct {
	PartitionSessionID PartitionSessionID
	Offsets            []OffsetRange
}

type PartitionCommittedOffset

type PartitionCommittedOffset struct {
	PartitionSessionID PartitionSessionID
	CommittedOffset    Offset
}

type PartitionData

type PartitionData struct {
	PartitionSessionID PartitionSessionID

	Batches []Batch
}

type PartitionSession

type PartitionSession struct {
	PartitionSessionID PartitionSessionID
	Path               string // Topic path of partition
	PartitionID        int64
}

type PartitionSessionID

type PartitionSessionID int64

func (*PartitionSessionID) FromInt64

func (id *PartitionSessionID) FromInt64(v int64)

func (PartitionSessionID) ToInt64

func (id PartitionSessionID) ToInt64() int64

type PartitionSessionStatusRequest

type PartitionSessionStatusRequest struct {
	PartitionSessionID PartitionSessionID
	// contains filtered or unexported fields
}

type PartitionSessionStatusResponse

type PartitionSessionStatusResponse struct {
	rawtopiccommon.ServerMessageMetadata

	PartitionSessionID     PartitionSessionID
	PartitionOffsets       OffsetRange
	WriteTimeHighWatermark time.Time
	// contains filtered or unexported fields
}

type ReadRequest

type ReadRequest struct {
	BytesSize int
	// contains filtered or unexported fields
}

type ReadResponse

type ReadResponse struct {
	rawtopiccommon.ServerMessageMetadata

	BytesSize     int
	PartitionData []PartitionData
	// contains filtered or unexported fields
}

func (*ReadResponse) GetBytesSize

func (r *ReadResponse) GetBytesSize() int

GetBytesSize implements trace.TopicReaderDataResponseInfo

func (*ReadResponse) GetPartitionBatchMessagesCounts

func (r *ReadResponse) GetPartitionBatchMessagesCounts() (partitionDataCount, batchCount, messagesCount int)

GetPartitionBatchMessagesCounts implements trace.TopicReaderDataResponseInfo

type ServerMessage

type ServerMessage interface {
	StatusData() rawtopiccommon.ServerMessageMetadata
	SetStatus(status rawydb.StatusCode)
	// contains filtered or unexported methods
}

type StartPartitionSessionRequest

type StartPartitionSessionRequest struct {
	rawtopiccommon.ServerMessageMetadata

	PartitionSession PartitionSession
	CommittedOffset  Offset
	PartitionOffsets OffsetRange
	// contains filtered or unexported fields
}

type StartPartitionSessionResponse

type StartPartitionSessionResponse struct {
	PartitionSessionID PartitionSessionID
	ReadOffset         OptionalOffset
	CommitOffset       OptionalOffset
	// contains filtered or unexported fields
}

type StopPartitionSessionRequest

type StopPartitionSessionRequest struct {
	rawtopiccommon.ServerMessageMetadata

	PartitionSessionID PartitionSessionID
	Graceful           bool
	CommittedOffset    Offset
	// contains filtered or unexported fields
}

type StopPartitionSessionResponse

type StopPartitionSessionResponse struct {
	PartitionSessionID PartitionSessionID
	// contains filtered or unexported fields
}

type StreamReader

type StreamReader struct {
	Stream GrpcStream
}

func (StreamReader) CloseSend

func (s StreamReader) CloseSend() error

func (StreamReader) Recv

func (s StreamReader) Recv() (ServerMessage, error)

func (StreamReader) Send

func (s StreamReader) Send(msg ClientMessage) (err error)

type TopicReadSettings

type TopicReadSettings struct {
	Path         string
	PartitionsID []int64

	MaxLag   rawoptional.Duration
	ReadFrom rawoptional.Time
}

type UpdateTokenRequest

type UpdateTokenRequest struct {
	rawtopiccommon.UpdateTokenRequest
	// contains filtered or unexported fields
}

type UpdateTokenResponse

type UpdateTokenResponse struct {
	rawtopiccommon.UpdateTokenResponse

	rawtopiccommon.ServerMessageMetadata
	// contains filtered or unexported fields
}