stream

Documentation

Index

• Constants
• func LoadServerAuthConfig()
• func ValidateStreamIteratorRequest(ctx context.Context, data []byte) error
• type CloseRecordsIteratorResponse
• type CreateRecordsIteratorResponse
• type GetRecordsIteratorStatsResponse
• type GetStreamRecordsResponse
• type LoginAccountResponse
• type LoginUserResponse
• type PutStreamRecordsResponse
• type RebuildStreamIndexResponse
• type Stream
  • func NewStream(info *StreamInfo, ingestBuffer *buffering.StreamIngestBuffer, ...) *Stream
  • func (s *Stream) AddIterator(it *StreamIterator) error
  • func (s *Stream) Close() error
  • func (s *Stream) CloseIterator(iterUUID StreamIteratorUUID) error
  • func (s *Stream) CloseIterators() error
  • func (s *Stream) GetInfo() *StreamInfo
  • func (s *Stream) GetIterator(iterUUID StreamIteratorUUID) (*StreamIterator, error)
  • func (s *Stream) GetProperties() *StreamProperties
  • func (s *Stream) GetRecords(c *fasthttp.RequestCtx, iterUUID StreamIteratorUUID, maxRecords uint) (*GetStreamRecordsResponse, error)
  • func (s *Stream) GetUUID() StreamUUID
  • func (s *Stream) Log()
  • func (s *Stream) MatchFilterProperties(jqFilter *gojq.Query) (bool, error)
  • func (s *Stream) PutMessage(c *fasthttp.RequestCtx, message map[string]interface{}) (MessageId, error)
  • func (s *Stream) PutMessages(c *fasthttp.RequestCtx, records []interface{}) ([]MessageId, error)
  • func (s *Stream) Run()
  • func (s *Stream) SetProperties(properties *StreamProperties)
  • func (s *Stream) Start() error
  • func (s *Stream) UpdateProperties(properties *StreamProperties)
• type StreamIterator
  • func NewStreamIterator(streamUUID StreamUUID, iteratorUUID StreamIteratorUUID, ...) (*StreamIterator, error)
  • func (it *StreamIterator) Close() error
  • func (it *StreamIterator) GetName() string
  • func (it *StreamIterator) GetRecords(c *fasthttp.RequestCtx, maxRecords uint) (*GetStreamRecordsResponse, error)
  • func (it *StreamIterator) GetUUID() StreamIteratorUUID
  • func (it *StreamIterator) Open() error
  • func (it *StreamIterator) SaveSeek() error
  • func (it *StreamIterator) Seek() error
• type StreamIteratorMap
• type StreamIteratorStats

Constants

const STREAM_STATE_NONE = 0

const STREAM_STATE_RUNNING = 2

const STREAM_STATE_STARTING = 1

const STREAM_STATE_STOPPING = 3

Functions

func LoadServerAuthConfig

func LoadServerAuthConfig()

func ValidateStreamIteratorRequest

func ValidateStreamIteratorRequest(ctx context.Context, data []byte) error

Types

type CloseRecordsIteratorResponse

type CloseRecordsIteratorResponse struct {
	Status             string             `json:"status"`
	Message            string             `json:"message"`
	StreamUUID         StreamUUID         `json:"streamUUID"`
	StreamIteratorUUID StreamIteratorUUID `json:"streamIteratorUUID"`
}

type CreateRecordsIteratorResponse

type CreateRecordsIteratorResponse struct {
	Status             string             `json:"status"`
	Message            string             `json:"message"`
	StreamUUID         StreamUUID         `json:"streamUUID"`
	StreamIteratorUUID StreamIteratorUUID `json:"streamIteratorUUID"`
}

type GetRecordsIteratorStatsResponse

type GetRecordsIteratorStatsResponse struct {
	Status             string             `json:"status"`
	Message            string             `json:"message"`
	StreamUUID         StreamUUID         `json:"streamUUID"`
	StreamIteratorUUID StreamIteratorUUID `json:"streamIteratorUUID"`
	LastRecordIdRead   MessageId          `json:"lastRecordIdRead"`
	Name               string             `json:"name"`
}

type GetStreamRecordsResponse

type GetStreamRecordsResponse struct {
	Status             string             `json:"status"`
	Duration           int64              `json:"duration"`
	Count              int64              `json:"count"`
	CountErrors        int64              `json:"countErrors"`
	CountSkipped       int64              `json:"countSkipped"`
	Remain             bool               `json:"remain"`
	LastRecordIdRead   MessageId          `json:"lastRecordIdRead"`
	StreamUUID         StreamUUID         `json:"streamUUID"`
	StreamIteratorUUID StreamIteratorUUID `json:"streamIteratorUUID"`
	Records            []interface{}      `json:"records"`
}

type LoginAccountResponse

type LoginAccountResponse struct {
	Status  string `json:"status"`
	Message string `json:"message"`
	JWT     string `json:"jwt"`
}

type LoginUserResponse

type LoginUserResponse struct {
	Status  string `json:"status"`
	Message string `json:"message"`
	JWT     string `json:"jwt"`
}

type PutStreamRecordsResponse

type PutStreamRecordsResponse struct {
	Status     string      `json:"status"`
	StreamUUID StreamUUID  `json:"streamUUID"`
	Duration   int64       `json:"duration"`
	Count      int64       `json:"count"`
	MessageIds []MessageId `json:"messageIds"`
}

type RebuildStreamIndexResponse

type RebuildStreamIndexResponse struct {
	Status     string      `json:"status"`
	Message    string      `json:"message"`
	StreamUUID StreamUUID  `json:"streamUUID"`
	Duration   int64       `json:"duration"`
	IndexStats interface{} `json:"indexStats"`
}

type Stream

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

func NewStream

func NewStream(info *StreamInfo, ingestBuffer *buffering.StreamIngestBuffer, logger *zap.Logger, logVerbosity int) *Stream

func (*Stream) AddIterator

func (s *Stream) AddIterator(it *StreamIterator) error

func (*Stream) Close

func (s *Stream) Close() error

func (*Stream) CloseIterator

func (s *Stream) CloseIterator(iterUUID StreamIteratorUUID) error

func (*Stream) CloseIterators

func (s *Stream) CloseIterators() error

func (*Stream) GetInfo

func (s *Stream) GetInfo() *StreamInfo

func (*Stream) GetIterator

func (s *Stream) GetIterator(iterUUID StreamIteratorUUID) (*StreamIterator, error)

func (*Stream) GetProperties

func (s *Stream) GetProperties() *StreamProperties

func (*Stream) GetRecords

func (s *Stream) GetRecords(c *fasthttp.RequestCtx, iterUUID StreamIteratorUUID, maxRecords uint) (*GetStreamRecordsResponse, error)

func (*Stream) GetUUID

func (s *Stream) GetUUID() StreamUUID

func (*Stream) Log

func (s *Stream) Log()

func (*Stream) MatchFilterProperties

func (s *Stream) MatchFilterProperties(jqFilter *gojq.Query) (bool, error)

func (*Stream) PutMessage

func (s *Stream) PutMessage(c *fasthttp.RequestCtx, message map[string]interface{}) (MessageId, error)

func (*Stream) PutMessages

func (s *Stream) PutMessages(c *fasthttp.RequestCtx, records []interface{}) ([]MessageId, error)

func (*Stream) Run

func (s *Stream) Run()

func (*Stream) SetProperties

func (s *Stream) SetProperties(properties *StreamProperties)

func (*Stream) Start

func (s *Stream) Start() error

func (*Stream) UpdateProperties

func (s *Stream) UpdateProperties(properties *StreamProperties)

type StreamIterator

type StreamIterator struct {
	LastRecordIdRead MessageId
	Stats            StreamIteratorStats
	// contains filtered or unexported fields
}

func NewStreamIterator

func NewStreamIterator(streamUUID StreamUUID, iteratorUUID StreamIteratorUUID, r *StreamIteratorRequest, handler IStreamIteratorHandler, logger *zap.Logger) (*StreamIterator, error)

func (*StreamIterator) Close

func (it *StreamIterator) Close() error

func (*StreamIterator) GetName

func (it *StreamIterator) GetName() string

func (*StreamIterator) GetRecords

func (it *StreamIterator) GetRecords(c *fasthttp.RequestCtx, maxRecords uint) (*GetStreamRecordsResponse, error)

func (*StreamIterator) GetUUID

func (it *StreamIterator) GetUUID() StreamIteratorUUID

func (*StreamIterator) Open

func (it *StreamIterator) Open() error

func (*StreamIterator) SaveSeek

func (it *StreamIterator) SaveSeek() error

func (*StreamIterator) Seek

func (it *StreamIterator) Seek() error

type StreamIteratorMap

type StreamIteratorMap = map[StreamIteratorUUID]*StreamIterator

type StreamIteratorStats

type StreamIteratorStats struct {
	BytesRead      int64
	RecordsRead    int64
	RecordsErrors  int64
	RecordsSkipped int64
	RecordsSent    int64
	LastTimeRead   time.Time
}

statistics of an iterator