fscache

README

fscache

Usage

Streaming File Cache for #golang

fscache allows multiple readers to read from a cache while its being written to. blog post

Using the Cache directly:

package main

import (
	"io"
	"log"
	"os"
	"time"

	"gopkg.in/djherbis/fscache.v0"
)

func main() {

	// create the cache, keys expire after 1 hour.
	c, err := fscache.New("./cache", 0755, time.Hour)
	if err != nil {
		log.Fatal(err.Error())
	}
	
	// wipe the cache when done
	defer c.Clean()

	// Get() and it's streams can be called concurrently but just for example:
	for i := 0; i < 3; i++ {
		r, w, err := c.Get("stream")
		if err != nil {
			log.Fatal(err.Error())
		}

		if w != nil { // a new stream, write to it.
			go func(){
				w.Write([]byte("hello world\n"))
				w.Close()
			}()
		}

		// the stream has started, read from it
		io.Copy(os.Stdout, r)
		r.Close()
	}
}

A Caching Middle-ware:

package main

import(
	"net/http"
	"time"

	"gopkg.in/djherbis/fscache.v0"
)

func main(){
	c, err := fscache.New("./cache", 0700, 0)
	if err != nil {
		log.Fatal(err.Error())
	}

	handler := func(w http.ResponseWriter, r *http.Request) {
		fmt.Fprintf(w, "%v: %s", time.Now(), "hello world")
	}

	http.ListenAndServe(":8080", fscache.Handler(c, http.HandlerFunc(handler)))
}

Installation

go get gopkg.in/djherbis/fscache.v0

Documentation

Overview

Example

// create the cache, keys expire after 1 hour.
c, err := New("./cache", 0755, time.Hour)
if err != nil {
	log.Fatal(err.Error())
}

// wipe the cache when done
defer c.Clean()

// Get() and it's streams can be called concurrently but just for example:
for i := 0; i < 3; i++ {
	r, w, err := c.Get("stream")
	if err != nil {
		log.Fatal(err.Error())
	}

	if w != nil { // a new stream, write to it.
		go func() {
			w.Write([]byte("hello world\n"))
			w.Close()
		}()
	}

	// the stream has started, read from it
	io.Copy(os.Stdout, r)
	r.Close()
}

Output:

hello world
hello world
hello world

Index

• func Handler(c Cache, h http.Handler) http.Handler
• func ListenAndServe(c Cache, addr string) error
• type Cache
  • func NewLayered(caches ...Cache) Cache
  • func NewPartition(d Distributor) Cache
  • func NewRemote(raddr string) Cache
• type CacheAccessor
• type CacheReader
  • func (r *CacheReader) Close() error
  • func (r *CacheReader) Size() (int64, bool, error)
• type Distributor
  • func NewDistributor(caches ...Cache) Distributor
• type Entry
  • func (e *Entry) InUse() bool
  • func (e *Entry) Name() string
• type FSCache
  • func New(dir string, perms os.FileMode, expiry time.Duration) (*FSCache, error)
  • func NewCache(fs FileSystem, grim Reaper) (*FSCache, error)
  • func NewCacheWithHaunter(fs FileSystem, haunter Haunter) (*FSCache, error)
  • func (c *FSCache) Clean() error
  • func (c *FSCache) Exists(key string) bool
  • func (c *FSCache) Get(key string) (r ReadAtCloser, w io.WriteCloser, err error)
  • func (c *FSCache) Remove(key string) error
• type FileInfo
  • func (f *FileInfo) AccessTime() time.Time
• type FileSystem
  • func NewFs(dir string, mode os.FileMode) (FileSystem, error)
  • func NewMemFs() FileSystem
• type FileSystemStater
• type Haunter
  • func NewLRUHaunterStrategy(haunter LRUHaunter) Haunter
  • func NewReaperHaunterStrategy(reaper Reaper) Haunter
• type LRUHaunter
  • func NewLRUHaunter(maxItems int, maxSize int64, period time.Duration) LRUHaunter
• type ReadAtCloser
• type Reaper
  • func NewReaper(expiry, period time.Duration) Reaper
• Bugs

Examples

• Package
• Handler

Functions

func Handler

func Handler(c Cache, h http.Handler) http.Handler

Handler is a caching middle-ware for http Handlers. It responds to http requests via the passed http.Handler, and caches the response using the passed cache. The cache key for the request is the req.URL.String(). Note: It does not cache http headers. It is more efficient to set them yourself.

Example

c, err := New("./server", 0700, 0)
if err != nil {
	log.Fatal(err.Error())
}
defer c.Clean()

handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintln(w, "Hello Client")
})

ts := httptest.NewServer(Handler(c, handler))
defer ts.Close()

resp, err := http.Get(ts.URL)
if err != nil {
	log.Fatal(err.Error())
}
io.Copy(os.Stdout, resp.Body)
resp.Body.Close()

Output:

Hello Client

func ListenAndServe

func ListenAndServe(c Cache, addr string) error

ListenAndServe hosts a Cache for access via NewRemote

Types

type Cache

type Cache interface {
	// Get manages access to the streams in the cache.
	// If the key does not exist, w != nil and you can start writing to the stream.
	// If the key does exist, w == nil.
	// r will always be non-nil as long as err == nil and you must close r when you're done reading.
	// Get can be called concurrently, and writing and reading is concurrent safe.
	Get(key string) (ReadAtCloser, io.WriteCloser, error)

	// Remove deletes the stream from the cache, blocking until the underlying
	// file can be deleted (all active streams finish with it).
	// It is safe to call Remove concurrently with Get.
	Remove(key string) error

	// Exists checks if a key is in the cache.
	// It is safe to call Exists concurrently with Get.
	Exists(key string) bool

	// Clean will empty the cache and delete the cache folder.
	// Clean is not safe to call while streams are being read/written.
	Clean() error
}

Cache works like a concurrent-safe map for streams.

func NewLayered

func NewLayered(caches ...Cache) Cache

NewLayered returns a Cache which stores its data in all the passed caches, when a key is requested it is loaded into all the caches above the first hit.

func NewPartition

func NewPartition(d Distributor) Cache

NewPartition returns a Cache which uses the Caches defined by the passed Distributor.

func NewRemote

func NewRemote(raddr string) Cache

NewRemote returns a Cache run via ListenAndServe

type CacheAccessor

type CacheAccessor interface {
	FileSystemStater
	EnumerateEntries(enumerator func(key string, e Entry) bool)
	RemoveFile(key string)
}

type CacheReader

type CacheReader struct {
	ReadAtCloser
	// contains filtered or unexported fields
}

func (*CacheReader) Close

func (r *CacheReader) Close() error

func (*CacheReader) Size

func (r *CacheReader) Size() (int64, bool, error)

Size returns the current size of the stream being read, the boolean it returns is true iff the stream is done being written (otherwise Size may change). An error is returned if the Size fails to be computed or is not supported by the underlying filesystem.

type Distributor

type Distributor interface {

	// GetCache will always return the same Cache for the same key.
	GetCache(key string) Cache

	// Clean should wipe all the caches this Distributor manages
	Clean() error
}

Distributor provides a way to partition keys into Caches.

func NewDistributor

func NewDistributor(caches ...Cache) Distributor

NewDistributor returns a Distributor which evenly distributes the keyspace into the passed caches.

type Entry

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

func (*Entry) InUse

func (e *Entry) InUse() bool

func (*Entry) Name

func (e *Entry) Name() string

type FSCache

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

func New

func New(dir string, perms os.FileMode, expiry time.Duration) (*FSCache, error)

New creates a new Cache using NewFs(dir, perms). expiry is the duration after which an un-accessed key will be removed from the cache, a zero value expiro means never expire.

func NewCache

func NewCache(fs FileSystem, grim Reaper) (*FSCache, error)

NewCache creates a new Cache based on FileSystem fs. fs.Files() are loaded using the name they were created with as a key. Reaper is used to determine when files expire, nil means never expire.

func NewCacheWithHaunter

func NewCacheWithHaunter(fs FileSystem, haunter Haunter) (*FSCache, error)

NewCacheWithHaunter create a new Cache based on FileSystem fs. fs.Files() are loaded using the name they were created with as a key. Haunter is used to determine when files expire, nil means never expire.

func (*FSCache) Clean

func (c *FSCache) Clean() error

func (*FSCache) Exists

func (c *FSCache) Exists(key string) bool

func (*FSCache) Get

func (c *FSCache) Get(key string) (r ReadAtCloser, w io.WriteCloser, err error)

func (*FSCache) Remove

func (c *FSCache) Remove(key string) error

type FileInfo

type FileInfo struct {
	os.FileInfo
	Atime time.Time
}

func (*FileInfo) AccessTime

func (f *FileInfo) AccessTime() time.Time

AccessTime returns the last time the file was read. It will be used to check expiry of a file, and must be concurrent safe with modifications to the FileSystem (writes, reads etc.)

type FileSystem

type FileSystem interface {
	// Stream FileSystem
	stream.FileSystem

	FileSystemStater

	// Reload should look through the FileSystem and call the supplied fn
	// with the key/filename pairs that are found.
	Reload(func(key, name string)) error

	// RemoveAll should empty the FileSystem of all files.
	RemoveAll() error
}

FileSystem is used as the source for a Cache.

func NewFs

func NewFs(dir string, mode os.FileMode) (FileSystem, error)

NewFs returns a FileSystem rooted at directory dir. Dir is created with perms if it doesn't exist.

func NewMemFs

func NewMemFs() FileSystem

NewMemFs creates an in-memory FileSystem. It does not support persistence (Reload is a nop).

type FileSystemStater

type FileSystemStater interface {
	// Stat takes a File.Name() and returns FileInfo interface
	Stat(name string) (FileInfo, error)
}

type Haunter

type Haunter interface {
	Haunt(c CacheAccessor)
	Next() time.Duration
}

func NewLRUHaunterStrategy

func NewLRUHaunterStrategy(haunter LRUHaunter) Haunter

NewLRUHaunterStrategy returns a simple scheduleHaunt which provides an implementation LRUHaunter strategy

func NewReaperHaunterStrategy

func NewReaperHaunterStrategy(reaper Reaper) Haunter

NewReaperHaunterStrategy returns a simple scheduleHaunt which provides an implementation Reaper strategy

type LRUHaunter

type LRUHaunter interface {
	// Returns the amount of time to wait before the next scheduled Reaping.
	Next() time.Duration

	// Given a CacheAccessor, return keys to reap list.
	Scrub(c CacheAccessor) []string
}

LRUHaunter is used to control when there are too many streams or the size of the streams is too big. It is called once right after loading, and then it is run again after every Next() period of time.

func NewLRUHaunter

func NewLRUHaunter(maxItems int, maxSize int64, period time.Duration) LRUHaunter

NewLRUHaunter returns a simple haunter which runs every "period" and scrubs older files when the total file size is over maxSize or total item count is over maxItems. If maxItems or maxSize are 0, they won't be checked

type ReadAtCloser

type ReadAtCloser interface {
	io.ReadCloser
	io.ReaderAt
}

ReadAtCloser is an io.ReadCloser, and an io.ReaderAt. It supports both so that Range Requests are possible.

type Reaper

type Reaper interface {
	// Returns the amount of time to wait before the next scheduled Reaping.
	Next() time.Duration

	// Given a key and the last r/w times of a file, return true
	// to remove the file from the cache, false to keep it.
	Reap(key string, lastRead, lastWrite time.Time) bool
}

Reaper is used to control when streams expire from the cache. It is called once right after loading, and then it is run again after every Next() period of time.

func NewReaper

func NewReaper(expiry, period time.Duration) Reaper

NewReaper returns a simple reaper which runs every "Period" and reaps files which are older than "expiry".

Notes

Bugs

• Return an error if cleaning fails