stream

README

stream

The library implements a simple streaming abstraction to store binary data and associated metadata at AWS storage services: AWS S3.

Inspiration

The library encourages developers to use Golang io.ReadCloser to abstract binary data and struct to define metadata. The library uses generic programming style to implement actual storage I/O, while expose metadata object as [T stream.Thing] with implicit conversion back and forth between a concrete struct(s). The library uses AWS Golang SDK under the hood.

Essentially, the library implement a following generic key-value trait to access domain objects.

type Stream[T stream.Thing] interface {
  Put(T, io.ReadCloser) error
  Get(T) (*T, io.ReadCloser, error)
  Remove(T) error
  Match(T) Seq[T]
}

Getting started

The library requires Go 1.18 or later due to usage of generics.

The latest version of the library is available at its main branch. All development, including new features and bug fixes, take place on the main branch using forking and pull requests as described in contribution guidelines. The stable version is available via Golang modules.

1. Use go get to retrieve the library and add it as dependency to your application.

go get -u github.com/fogfish/stream

1. Import required package in your code

import (
  "github.com/fogfish/stream"
  "github.com/fogfish/stream/creek"
)

Data types definition

Data types definition is an essential part of development with stream library. Golang structs declares metadata of your binary streams. Public fields are serialized into S3 metadata attributes, the field tag metadata controls marshal/unmarshal process.

The library demands from each structure implementation of Thing interface. This type acts as struct annotation -- Golang compiler raises an error at compile time if other data type is supplied for Stream I/O. Secondly, each structure defines unique "composite primary key". The library encourages definition of both partition and sort keys, which facilitates linked-data, hierarchical structures and cheap relations between data elements.

import "github.com/fogfish/stream"

type Note struct {
	Author string `metadata:"Author"`
	ID     string `metadata:"Id"`
}

//
// Identity implements thing interface
func (n Note) HashKey() string { return n.Author }
func (n Note) SortKey() string { return n.ID }

//
// this data type is a normal Golang struct
// just create an instance, fill required fields
var note := Note{
  Author:  "haskell",
  ID:      "8980789222",
}

Stream I/O

Please see and try examples. Its cover all basic use-cases with runnable code snippets.

go run examples/stream/stream.go s3:///my-bucket

The following code snippet shows a typical I/O patterns

import (
  "github.com/fogfish/stream"
  "github.com/fogfish/stream/creek"
)

//
// Create client and bind it with the bucket
// Use URI notation to specify the diver (s3://) and the bucket (/my-bucket) 
db := creek.Must(creek.New[Note]("s3:///my-bucket"))

//
// Write the stream with Put
stream := io.NopCloser(strings.NewReader("..."))
if err := db.Put(note, stream); err != nil {
}

//
// Lookup the stream using Get. This function takes input structure as key
// and return a new copy upon the completion. The only requirement - ID has to
// be defined.
note, stream, err := db.Get(
  Note{
    Author:  "haskell",
    ID:      "8980789222",
  },
)

switch {
case nil:
  // success
case recoverNotFound(err):
  // not found
default:
  // other i/o error
}

//
// Remove the stream using Remove
err := db.Remove(
  Note{
    Author:  "haskell",
    ID:      "8980789222",
  },
)

if err != nil { /* ... */ }

Working with streams metadata

Please see the original AWS post about Working with object metadata. The library support both system and user-defined metadata. System define metadata is few well-known attributes: Cache-Control, Content-Encoding, Content-Language, Content-Type and Expires.

type Note struct {
  // User-defined metadata
  Author    string `metadata:"Author"`
  ID        string `metadata:"Id"`
  Custom    string `metadata:"Custom"`
  Attribute string `metadata:"Attribute"`
  // System metadata
  CacheControl    string    `metadata:"Cache-Control"`
  ContentEncoding string    `metadata:"Content-Encoding"`
  ContentLanguage string    `metadata:"Content-Language"`
  ContentType     string    `metadata:"Content-Type"`
  Expires         time.Time `metadata:"Expires"`
}

Error Handling

The library enforces for "assert errors for behavior, not type" as the error handling strategy, see the post for details.

Use following behaviors to recover from errors

type ErrorCode interface{ ErrorCode() string }

type NotFound interface { NotFound() string }

How To Contribute

The library is MIT licensed and accepts contributions via GitHub pull requests:

1. Fork it
2. Create your feature branch (git checkout -b my-new-feature)
3. Commit your changes (git commit -am 'Added some feature')
4. Push to the branch (git push origin my-new-feature)
5. Create new Pull Request

The build and testing process requires Go version 1.13 or later.

build and test library.

git clone https://github.com/fogfish/dynamo
cd dynamo
go test

commit message

The commit message helps us to write a good release note, speed-up review process. The message should address two question what changed and why. The project follows the template defined by chapter Contributing to a Project of Git book.

bugs

If you experience any issues with the library, please let us know via GitHub issues. We appreciate detailed and accurate reports that help us to identity and replicate the issue.

License

Documentation

Index

• type Config
  • func NewConfig(opts ...Option) (*Config, error)
• type Option
  • func WithAwsConfig(aws aws.Config) Option
  • func WithPrefixes(prefixes curie.Prefixes) Option
  • func WithURI(uri string) Option
• type Seq
• type SeqConfig
• type SeqLazy
• type Stream
• type StreamGetter
• type StreamGetterNoContext
• type StreamNoContext
• type StreamPattern
• type StreamPatternNoContext
• type StreamReader
• type StreamReaderNoContext
• type StreamWriter
• type StreamWriterNoContext
• type Thing
• type URL
  • func (uri *URL) Query(key, def string) string
  • func (uri *URL) Segments() []string
  • func (uri *URL) String() string

Types

type Config

type Config struct {
	URI      *URL
	Prefixes curie.Prefixes
	AWS      aws.Config
}

Config options for the connection

func NewConfig

func NewConfig(opts ...Option) (*Config, error)

NewConfig creates Config with default options

type Option

type Option func(cfg *Config) error

Option type to configure the connection

func WithAwsConfig

func WithAwsConfig(aws aws.Config) Option

WithSession defines AWS I/O Session to be used in the context

func WithPrefixes

func WithPrefixes(prefixes curie.Prefixes) Option

WithPrefixes defines prefixes for CURIEs

func WithURI

func WithURI(uri string) Option

WithURI defines destination URI

type Seq

type Seq[T Thing] interface {
	SeqLazy[T]
	SeqConfig[T]

	// Sequence transformer
	FMap(func(T, io.ReadCloser) error) error
}

Seq is an interface to transform collection of objects

db.Match(...).FMap(func(key Thing, val io.ReadCloser) error { ... })

type SeqConfig

type SeqConfig[T Thing] interface {
	// Limit sequence size to N elements (pagination)
	Limit(int64) Seq[T]
	// Continue limited sequence from the cursor
	Continue(Thing) Seq[T]
	// Reverse order of sequence
	Reverse() Seq[T]
}

SeqConfig configures optional sequence behavior

type SeqLazy

type SeqLazy[T Thing] interface {
	// Head lifts first element of sequence
	Head() (T, io.ReadCloser, error)
	// Tail evaluates tail of sequence
	Tail() bool
	// Error returns error of stream evaluation
	Error() error
	// Cursor is the global position in the sequence
	Cursor() Thing
}

SeqLazy is an interface to iterate through collection of objects at storage

type Stream

type Stream[T Thing] interface {
	StreamReader[T]
	StreamWriter[T]
}

Stream is a generic key-value trait to access domain objects.

type StreamGetter

type StreamGetter[T Thing] interface {
	Has(context.Context, T) (bool, error)
	URL(context.Context, T, time.Duration) (string, error)
	Get(context.Context, T) (T, io.ReadCloser, error)
}

StreamGetter defines read by key notation

type StreamGetterNoContext

type StreamGetterNoContext[T Thing] interface {
	Has(T) (bool, error)
	URL(T, time.Duration) (string, error)
	Get(T) (T, io.ReadCloser, error)
}

StreamGetterNoContext defines read by key notation

type StreamNoContext

type StreamNoContext[T Thing] interface {
	StreamReaderNoContext[T]
	StreamWriterNoContext[T]
}

StreamNoContext is a generic key-value trait to access domain objects.

type StreamPattern

type StreamPattern[T Thing] interface {
	Match(context.Context, T) Seq[T]
}

StreamPattern defines simple pattern matching lookup I/O

type StreamPatternNoContext

type StreamPatternNoContext[T Thing] interface {
	Match(T) Seq[T]
}

StreamPatternNoContext defines simple pattern matching lookup I/O

type StreamReader

type StreamReader[T Thing] interface {
	StreamGetter[T]
	StreamPattern[T]
}

KeyValReader a generic key-value trait to read domain objects

type StreamReaderNoContext

type StreamReaderNoContext[T Thing] interface {
	StreamGetterNoContext[T]
	StreamPatternNoContext[T]
}

StreamReaderNoContext a generic key-value trait to read domain objects

type StreamWriter

type StreamWriter[T Thing] interface {
	Put(context.Context, T, io.ReadCloser) error
	Copy(context.Context, T, T) error
	Remove(context.Context, T) error
}

StreamWriter defines a generic key-value writer

type StreamWriterNoContext

type StreamWriterNoContext[T Thing] interface {
	Put(T, io.ReadCloser) error
	Copy(T, T) error
	Remove(T) error
}

StreamWriterNoContext defines a generic key-value writer

type Thing

type Thing interface {
	HashKey() curie.IRI
	SortKey() curie.IRI
}

Thing is the most generic item type used by the library to abstract writable/readable streams into storage services.

The interfaces declares anything that have a unique identifier. The unique identity is exposed by pair of string: HashKey and SortKey.

type URL

type URL url.URL

URL custom type with helper functions

func (*URL) Query

func (uri *URL) Query(key, def string) string

query parameters

func (*URL) Segments

func (uri *URL) Segments() []string

path segments of length

func (*URL) String

func (uri *URL) String() string