streaming_data

Documentation

Index

• type Encoder
  • func (e *Encoder[TK]) Close() error
  • func (e *Encoder[TK]) Encode(v any) error
  • func (e *Encoder[TK]) SetEscapeHTML(on bool)
  • func (e *Encoder[TK]) SetIndent(prefix, indent string)
• type StreamingDataKey
  • func (x StreamingDataKey[TK]) Compare(other interface{}) int
• type StreamingDataStore
  • func NewStreamingDataStore[TK btree.Comparable](ctx context.Context, name string, trans in_red_ck.Transaction) (*StreamingDataStore[TK], error)
  • func OpenStreamingDataStore[TK btree.Comparable](ctx context.Context, name string, trans in_red_ck.Transaction) (*StreamingDataStore[TK], error)
  • func (s *StreamingDataStore[TK]) Add(ctx context.Context, key TK) (*Encoder[TK], error)
  • func (s *StreamingDataStore[TK]) Count() int64
  • func (s *StreamingDataStore[TK]) FindChunk(ctx context.Context, key TK, chunkIndex int) (bool, error)
  • func (s *StreamingDataStore[TK]) FindOne(ctx context.Context, key TK) (bool, error)
  • func (s *StreamingDataStore[TK]) First(ctx context.Context) (bool, error)
  • func (s *StreamingDataStore[TK]) GetCurrentKey() TK
  • func (s *StreamingDataStore[TK]) GetCurrentValue(ctx context.Context) (*json.Decoder, error)
  • func (s *StreamingDataStore[TK]) IsUnique() bool
  • func (s *StreamingDataStore[TK]) Last(ctx context.Context) (bool, error)
  • func (s *StreamingDataStore[TK]) Next(ctx context.Context) (bool, error)
  • func (s *StreamingDataStore[TK]) Previous(ctx context.Context) (bool, error)
  • func (s *StreamingDataStore[TK]) Remove(ctx context.Context, key TK) (bool, error)
  • func (s *StreamingDataStore[TK]) RemoveCurrentItem(ctx context.Context) (bool, error)
  • func (s *StreamingDataStore[TK]) Update(ctx context.Context, key TK) (*Encoder[TK], error)
  • func (s *StreamingDataStore[TK]) UpdateCurrentItem(ctx context.Context) (*Encoder[TK], error)

Types

type Encoder

type Encoder[TK btree.Comparable] struct {
	// contains filtered or unexported fields
}

An Encoder writes JSON values to an output stream by delegating to JSON Encoder.

func (*Encoder[TK]) Close

func (e *Encoder[TK]) Close() error

Close is only useful for Update/UpdateCurrentItem. It allows StreamingDataStore to do any house cleanup if needed. Not necessary for Add/AddIfNotExists methods of StreamingDataStore.

Example, on Update/UpdateCurrentItem, store will ensure to cleanup or delete any chunks that were not replaced by the encoder/writer.

func (*Encoder[TK]) Encode

func (e *Encoder[TK]) Encode(v any) error

Encode writes the JSON encoding of v to the stream, followed by a newline character.

See the documentation for Marshal for details about the conversion of Go values to JSON.

func (*Encoder[TK]) SetEscapeHTML

func (e *Encoder[TK]) SetEscapeHTML(on bool)

SetEscapeHTML specifies whether problematic HTML characters should be escaped inside JSON quoted strings. The default behavior is to escape &, <, and > to \u0026, \u003c, and \u003e to avoid certain safety problems that can arise when embedding JSON in HTML.

In non-HTML settings where the escaping interferes with the readability of the output, SetEscapeHTML(false) disables this behavior.

func (*Encoder[TK]) SetIndent

func (e *Encoder[TK]) SetIndent(prefix, indent string)

SetIndent instructs the encoder to format each subsequent encoded value as if indented by the package-level function Indent(dst, src, prefix, indent). Calling SetIndent("", "") disables indentation.

type StreamingDataKey

type StreamingDataKey[TK btree.Comparable] struct {
	Key        TK
	ChunkIndex int
}

StreamingDataKey is the Key struct for our Streaming Data Store. Take note, it has to be "public"(starts with capital letter) and member fields "public" as well so it can get persisted by JSON encoder/decoder properly.

func (StreamingDataKey[TK]) Compare

func (x StreamingDataKey[TK]) Compare(other interface{}) int

Compare is our Streaming Data Store comparer of keys.

type StreamingDataStore

type StreamingDataStore[TK btree.Comparable] struct {
	// contains filtered or unexported fields
}

StreamingDataStore contains methods useful for storage & management of entries that allow encoding and decoding to/from data streams.

func NewStreamingDataStore

func NewStreamingDataStore[TK btree.Comparable](ctx context.Context, name string, trans in_red_ck.Transaction) (*StreamingDataStore[TK], error)

NewStreamingDataStore instantiates a new Data Store for use in "streaming data". That is, the "value" is saved in separate segment(partition in Cassandra) & actively persisted to the backend, e.g. - call to Add method will save right away to the separate segment and on commit, it will be a quick action as data is already saved to the data segments.

This behaviour makes this store ideal for data management of huge blobs, like movies or huge data graphs.

func OpenStreamingDataStore

func OpenStreamingDataStore[TK btree.Comparable](ctx context.Context, name string, trans in_red_ck.Transaction) (*StreamingDataStore[TK], error)

OpenStreamingDataStore opens an existing data store for use in "streaming data".

func (*StreamingDataStore[TK]) Add

func (s *StreamingDataStore[TK]) Add(ctx context.Context, key TK) (*Encoder[TK], error)

Add insert an item to the b-tree and returns an encoder you can use to write the streaming data on.

func (*StreamingDataStore[TK]) Count

func (s *StreamingDataStore[TK]) Count() int64

Returns the total number of data chunks in this store.

func (*StreamingDataStore[TK]) FindChunk

func (s *StreamingDataStore[TK]) FindChunk(ctx context.Context, key TK, chunkIndex int) (bool, error)

FindChunk will search Btree for an item with a given key and chunkIndex. If you passed in a chunkIndex that is beyond the number of chunks of the item then it will return false.

You can use FindChunk or FindOne & Next to navigate to the fragment or chunk # you are targeting to download.

func (*StreamingDataStore[TK]) FindOne

func (s *StreamingDataStore[TK]) FindOne(ctx context.Context, key TK) (bool, error)

FindOne will search Btree for an item with a given key. Return true if found, otherwise false. Use the CurrentKey/CurrentValue to retrieve the "current item" details(key &/or decoder).

func (*StreamingDataStore[TK]) First

func (s *StreamingDataStore[TK]) First(ctx context.Context) (bool, error)

First positions the "cursor" to the first item as per key ordering. Use the CurrentKey/CurrentValue to retrieve the "current item" details(key &/or value).

func (*StreamingDataStore[TK]) GetCurrentKey

func (s *StreamingDataStore[TK]) GetCurrentKey() TK

GetCurrentKey returns the current item's key.

func (*StreamingDataStore[TK]) GetCurrentValue

func (s *StreamingDataStore[TK]) GetCurrentValue(ctx context.Context) (*json.Decoder, error)

GetCurrentValue returns the current item's decoder you can use to download the data chunks (or stream it down).

func (*StreamingDataStore[TK]) IsUnique

func (s *StreamingDataStore[TK]) IsUnique() bool

IsUnique always returns true for Streaming Data Store.

func (*StreamingDataStore[TK]) Last

func (s *StreamingDataStore[TK]) Last(ctx context.Context) (bool, error)

Last positionts the "cursor" to the last item as per key ordering. Use the CurrentKey/CurrentValue to retrieve the "current item" details(key &/or value).

func (*StreamingDataStore[TK]) Next

func (s *StreamingDataStore[TK]) Next(ctx context.Context) (bool, error)

Next positions the "cursor" to the next item chunk as per key ordering. Use the CurrentKey/CurrentValue to retrieve the "current item" details(key &/or value).

Ensure you are not navigating passed the target chunk via calling GetCurrentKey and checking that it is still the Key of the item you are interested about.

func (*StreamingDataStore[TK]) Previous

func (s *StreamingDataStore[TK]) Previous(ctx context.Context) (bool, error)

Previous positions the "cursor" to the previous item chunk as per key ordering. Use the CurrentKey/CurrentValue to retrieve the "current item" details(key &/or value).

func (*StreamingDataStore[TK]) Remove

func (s *StreamingDataStore[TK]) Remove(ctx context.Context, key TK) (bool, error)

Remove will delete the item's data chunks given its key.

func (*StreamingDataStore[TK]) RemoveCurrentItem

func (s *StreamingDataStore[TK]) RemoveCurrentItem(ctx context.Context) (bool, error)

RemoveCurrentItem will delete the current item's data chunks.

func (*StreamingDataStore[TK]) Update

func (s *StreamingDataStore[TK]) Update(ctx context.Context, key TK) (*Encoder[TK], error)

Update finds the item with key and returns an encoder you can use to upload and update the item's data chunks.

func (*StreamingDataStore[TK]) UpdateCurrentItem

func (s *StreamingDataStore[TK]) UpdateCurrentItem(ctx context.Context) (*Encoder[TK], error)

UpdateCurrentItem will return an encoder that will allow you to update the current item's data chunks.