fix http writes without deflate and clean up APIs

Author: bakape

cache.go

@@ -93,15 +93,15 @@ func (c *Cache) NewFrontend(get Getter) *Frontend {
 
 // Get or create a new record in the cache.
 // fresh=true, if record is freshly created and requires population.
-func (c *Cache) getRecord(loc recordLocation) (rec *record, fresh bool) {
+func (c *Cache) getRecord(loc recordLocation) (rec *Record, fresh bool) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 
 	recWithMeta, ok := c.record(loc)
 	if !ok {
 		recWithMeta = recordWithMeta{
 			node: c.lruList.Prepend(loc),
-			rec:  new(record),
+			rec:  new(Record),
 		}
 		recWithMeta.rec.semaphore.Init() // Block all reads until population
 	} else {
@@ -148,7 +148,7 @@ func (c *Cache) record(loc recordLocation) (recordWithMeta, bool) {
 }
 
 // Set record used memory
-func (c *Cache) setUsedMemory(src *record, loc recordLocation, memoryUsed int) {
+func (c *Cache) setUsedMemory(src *Record, loc recordLocation, memoryUsed int) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 

components.go

@@ -1,6 +1,8 @@
 package recache
 
 import (
+	"bytes"
+	"compress/flate"
 	"crypto/sha1"
 	"io"
 )
@@ -12,6 +14,7 @@ type component interface {
 	Size() int
 	Hash() [sha1.Size]byte
 	GetFrameDescriptor() frameDescriptor
+	Decompress() io.Reader
 }
 
 // Common part of both buffer and reference components
@@ -36,9 +39,7 @@ func (b buffer) WriteTo(w io.Writer) (int64, error) {
 }
 
 func (b buffer) NewReader() io.Reader {
-	return &bufferReader{
-		buffer: b,
-	}
+	return bytes.NewReader(b.data)
 }
 
 func (b buffer) Size() int {
@@ -49,28 +50,15 @@ func (b buffer) GetFrameDescriptor() frameDescriptor {
 	return b.frameDescriptor
 }
 
-// Adapter for reading data from component w/o mutating it
-type bufferReader struct {
-	off int
-	buffer
-}
-
-func (r *bufferReader) Read(p []byte) (n int, err error) {
-	if len(p) == 0 {
-		return
-	}
-	if r.off >= len(r.data) {
-		return 0, io.EOF
-	}
-	n = copy(p, r.data[r.off:])
-	r.off += n
-	return
+// Read component as decompressed stream
+func (b buffer) Decompress() io.Reader {
+	return flate.NewReader(b.NewReader())
 }
 
 // Reference to another record
 type recordReference struct {
 	componentCommon
-	*record
+	*Record
 }
 
 func (r recordReference) Size() int {

frontend.go

@@ -1,7 +1,6 @@
 package recache
 
 import (
-	"compress/flate"
 	"crypto/sha1"
 	"encoding/base64"
 	"encoding/binary"
@@ -26,36 +25,6 @@ type Key interface{}
 // Getter must be thread-safe.
 type Getter func(Key, *RecordWriter) error
 
-// Readable stream with support for io.WriterTo and conversion to
-// io.Reader interfaces
-type Streamer interface {
-	// Can be called safely from multiple goroutines
-	io.WriterTo
-
-	// Create a new io.Reader for this stream.
-	// Multiple instances of such an io.Reader can exist and be read
-	// concurrently.
-	NewReader() io.Reader
-
-	// Convenience method for efficiently decoding stream contents as JSON into
-	// the destination variable.
-	//
-	// dst: pointer to destination variable
-	DecodeJSON(dst interface{}) error
-
-	// Create a new io.ReadCloser for the Decompressped content of this stream.
-	//
-	// It is the caller's responsibility to call Close on the io.ReadCloser
-	// when finished reading.
-	Decompress() io.Reader
-
-	// Return SHA1 hash of the content
-	SHA1() [sha1.Size]byte
-
-	// Return strong etag of content
-	ETag() string
-}
-
 // A frontend for accessing the cache contents
 type Frontend struct {
 	id     int
@@ -64,7 +33,7 @@ type Frontend struct {
 }
 
 // Populates a record using the registered Getter
-func (f *Frontend) populate(k Key, rec *record) (err error) {
+func (f *Frontend) populate(k Key, rec *Record) (err error) {
 	rw := RecordWriter{
 		cache:    f.cache.id,
 		frontend: f.id,
@@ -125,7 +94,7 @@ func (f *Frontend) populate(k Key, rec *record) (err error) {
 }
 
 // Get a record by key and block until it has been generated
-func (f *Frontend) getGeneratedRecord(k Key) (rec *record, err error) {
+func (f *Frontend) getGeneratedRecord(k Key) (rec *Record, err error) {
 	loc := recordLocation{f.id, k}
 	rec, fresh := f.cache.getRecord(loc)
 	if fresh {
@@ -151,19 +120,15 @@ func (f *Frontend) getGeneratedRecord(k Key) (rec *record, err error) {
 	return
 }
 
-// Retrieve or generate data by key and return a consumable result Stream
-func (f *Frontend) Get(k Key) (s Streamer, err error) {
-	rec, err := f.getGeneratedRecord(k)
-	if err != nil {
-		return
-	}
-	s = recordDecoder{rec}
-	return
+// Retrieve or generate data by key and return cache Record
+func (f *Frontend) Get(k Key) (*Record, error) {
+	return f.getGeneratedRecord(k)
 }
 
 // Retrieve or generate data by key and write it to w.
 // Writes ETag to w and returns 304 on ETag match without writing data.
-// Sets "Content-Encoding" header to "deflate".
+// Sets "Content-Encoding" header to "deflate", if client support deflate
+// compressions
 func (f *Frontend) WriteHTTP(k Key, w http.ResponseWriter, r *http.Request,
 ) (n int64, err error) {
 	rec, err := f.getGeneratedRecord(k)
@@ -180,14 +145,14 @@ func (f *Frontend) WriteHTTP(k Key, w http.ResponseWriter, r *http.Request,
 	if !supportsDeflate {
 		// Different eTag to maintain strong eTag byte-equivalence guarantee by
 		// differing it from the compressed eTag.
-		eTag = eTag[:len(eTag)-2] + `-uc"`
+		eTag = rec.ETagDecompressed()
 	}
 	if r.Header.Get("If-None-Match") == eTag {
 		w.WriteHeader(304)
 		return
 	}
 	h := w.Header()
-	h.Set("ETag", rec.eTag)
+	h.Set("ETag", eTag)
 
 	if supportsDeflate {
 		// If client accepts deflate compression use efficient deflate stream
@@ -242,15 +207,9 @@ func (f *Frontend) WriteHTTP(k Key, w http.ResponseWriter, r *http.Request,
 		}
 		n += 6
 	} else {
-		// Streaming decompression for clients that don't accept deflate
+		// Streaming decompression for clients that don't support deflate
 		// compression
-
-		dr := flate.NewReader(rec.NewReader())
-		n, err = io.Copy(w, dr)
-		if err != nil {
-			return
-		}
-		err = dr.Close()
+		n, err = io.Copy(w, rec.Decompress())
 	}
 
 	return

frontend_test.go

@@ -173,7 +173,7 @@ func TestWriteHTTP(t *testing.T) {
 	for i := 0; i < 3; i++ {
 		c := cases[i]
 		c.name += " no deflate"
-		c.useDeflate = true
+		c.useDeflate = false
 		cases = append(cases, c)
 	}
 
@@ -204,7 +204,13 @@ func TestWriteHTTP(t *testing.T) {
 				if err != nil {
 					t.Fatal(err)
 				}
-				assertEquals(t, s.ETag(), etag)
+				var stdETag string
+				if c.useDeflate {
+					stdETag = s.ETag()
+				} else {
+					stdETag = s.ETagDecompressed()
+				}
+				assertEquals(t, stdETag, etag)
 
 				body := rec.Body
 				if c.useDeflate {

go.mod

@@ -1,3 +1,3 @@
-module github.com/bakape/recache/v5
+module github.com/bakape/recache/v6
 
 go 1.13

record.go

@@ -41,11 +41,11 @@ type recordWithMeta struct {
 	// lock on the cache mutex held.
 	//
 	// Record must be a pointer, because it contains mutexes.
-	rec *record
+	rec *Record
 }
 
 // Data storage unit in the cache. Linked to a single Key on a Frontend.
-type record struct {
+type Record struct {
 	semaphore semaphore
 
 	// Contained data and metainformation
@@ -68,7 +68,8 @@ type componentNode struct {
 	next *componentNode
 }
 
-func (r *record) WriteTo(w io.Writer) (n int64, err error) {
+// Implements io.WrWriteTo
+func (r *Record) WriteTo(w io.Writer) (n int64, err error) {
 	for c, m := &r.data, int64(0); c != nil; c = c.next {
 		m, err = c.WriteTo(w)
 		if err != nil {
@@ -79,19 +80,51 @@ func (r *record) WriteTo(w io.Writer) (n int64, err error) {
 	return
 }
 
-func (r *record) NewReader() io.Reader {
+// Create a new io.Reader for this stream.
+// Multiple instances of such an io.Reader can exist and be read
+// concurrently.
+func (r *Record) NewReader() io.Reader {
 	return &recordReader{
-		next:    r.data.next,
 		current: r.data.NewReader(),
-		record:  r,
+		next:    r.data.next,
 	}
 }
 
+// Create a new io.ReadCloser for the Decompressped content of this stream.
+//
+// It is the caller's responsibility to call Close on the io.ReadCloser
+// when finished reading.
+func (r *Record) Decompress() io.Reader {
+	return eofCaster{flate.NewReader(r.NewReader())}
+}
+
+// Convenience method for efficiently decoding stream contents as JSON into
+// the destination variable.
+//
+// dst: pointer to destination variable
+func (r *Record) DecodeJSON(dst interface{}) (err error) {
+	return json.NewDecoder(r.Decompress()).Decode(dst)
+}
+
+// Return SHA1 hash of the content
+func (r *Record) SHA1() [sha1.Size]byte {
+	return r.hash
+}
+
+// Return strong ETag of content, if served as a compressed stream
+func (r *Record) ETag() string {
+	return r.eTag
+}
+
+// Return strong ETag of content, if served as a decompressed stream
+func (r *Record) ETagDecompressed() string {
+	return r.eTag[:len(r.eTag)-1] + `-uc"`
+}
+
 // Adapter for reading data from record w/o mutating it
 type recordReader struct {
 	current io.Reader
 	next    *componentNode
-	*record
 }
 
 func (r *recordReader) Read(p []byte) (n int, err error) {
@@ -116,23 +149,16 @@ func (r *recordReader) Read(p []byte) (n int, err error) {
 	return
 }
 
-// Adapter, that enables decoding the record as JSON
-type recordDecoder struct {
-	*record
-}
-
-func (r recordDecoder) DecodeJSON(dst interface{}) (err error) {
-	return json.NewDecoder(r.Decompress()).Decode(dst)
-}
-
-func (r recordDecoder) Decompress() io.Reader {
-	return flate.NewReader(r.NewReader())
-}
-
-func (r recordDecoder) SHA1() [sha1.Size]byte {
-	return r.record.hash
+// Suppresses unexpected EOF errors resulting as a consequence of flate using
+// bufio
+type eofCaster struct {
+	io.Reader
 }
 
-func (r recordDecoder) ETag() string {
-	return r.eTag
+func (e eofCaster) Read(p []byte) (n int, err error) {
+	n, err = e.Reader.Read(p)
+	if err == io.ErrUnexpectedEOF {
+		err = io.EOF
+	}
+	return
 }

utils_test.go

@@ -21,7 +21,7 @@ func assertEquals(t *testing.T, res, std interface{}) {
 	}
 }
 
-func decodeJSON(t *testing.T, src Streamer, dst interface{}) {
+func decodeJSON(t *testing.T, src *Record, dst interface{}) {
 	t.Helper()
 
 	err := src.DecodeJSON(&dst)
@@ -30,7 +30,7 @@ func decodeJSON(t *testing.T, src Streamer, dst interface{}) {
 	}
 }
 
-func assertJsonStringEquals(t *testing.T, src Streamer, std string) {
+func assertJsonStringEquals(t *testing.T, src *Record, std string) {
 	t.Helper()
 
 	var res string