stream

README

go-stream

go-stream will allow stream reading/writing on Storage via adding a new stream layer.

This lib will maintain data between under and upper storage layer.

Status

This project is a Proof of Concept, please connect with us via issues if you have interest on it.

Usage

Init a stream

Init a stream via upper and under storage. All data will be written into upper first. go-stream will persist data from upper to under async.

s, err := stream.NewWithConfig(&stream.Config{
    Upper:         upperStore,
    Under:         underStore,
    // Set speed limit here to prevent upper been written too fast.
    SpeedLimit:    10 * 1024 * 1024,
    PersistMethod: stream.PersistMethodMultipart,
})
if err != nil {
	return err
}
go s.Serve()
go func() {
    for v := range s.Errors() {
        log.Printf("got error: %v", v)
    }
}()

Start a new branch

br, err := s.StartBranch(id, name)
if err != nil {
	return err
}

Write data into a branch

• Write

Data should be written in order.

n, err := br.Write(uint64(i), bs)
if err != nil {
    t.Fatal(err)
}

• ReadFrom

Data will be read from the Reader until io.EOF.

n, err := br.ReadFrom(r)
if err != nil {
    t.Fatal(err)
}

Complete a branch

Call Complete to finish a branch write job. Only after this call, the written data is persisted.

err = br.Complete()
if err != nil {
    t.Fatal(err)
}

Documentation

Index

• Constants
• type Branch
  • func (br *Branch) Complete() (err error)
  • func (br *Branch) ReadFrom(r io.Reader) (n int64, err error)
  • func (br *Branch) Write(idx uint64, data []byte) (n int64, err error)
• type Config
• type Stream
  • func New(upper, under types.Storager) (s *Stream, err error)
  • func NewWithConfig(cfg *Config) (s *Stream, err error)
  • func (s *Stream) Errors() chan error
  • func (s *Stream) Serve()
  • func (s *Stream) StartBranch(id uint64, path string) (br *Branch, err error)

Constants

const (
	PersistMethodWrite     = "write"
	PersistMethodMultipart = "multipart"
	PersistMethodAppend    = "append"
)

Types

type Branch

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

func (*Branch) Complete

func (br *Branch) Complete() (err error)

func (*Branch) ReadFrom

func (br *Branch) ReadFrom(r io.Reader) (n int64, err error)

func (*Branch) Write

func (br *Branch) Write(idx uint64, data []byte) (n int64, err error)

type Config

type Config struct {
	Upper types.Storager
	Under types.Storager

	// SpeedLimit for upper storager, unit is B/s
	SpeedLimit    int
	PersistMethod string
}

type Stream

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

func New

func New(upper, under types.Storager) (s *Stream, err error)

func NewWithConfig

func NewWithConfig(cfg *Config) (s *Stream, err error)

func (*Stream) Errors

func (s *Stream) Errors() chan error

func (*Stream) Serve

func (s *Stream) Serve()

func (*Stream) StartBranch

func (s *Stream) StartBranch(id uint64, path string) (br *Branch, err error)