// (1 << logWindowSize) - 1.
windowMask int
+ eof bool // has eof been reached on input?
+ sync bool // writer wants to flush
+ syncChan chan os.Error
+
// hashHead[hashValue] contains the largest inputIndex with the specified hash value
hashHead []int
}
func (d *compressor) fillWindow(index int) (int, os.Error) {
+ if d.sync {
+ return index, nil
+ }
wSize := d.windowMask + 1
if index >= wSize+wSize-(minMatchLength+maxMatchLength) {
// shift the window by wSize
d.hashPrev[i] = max(h-wSize, -1)
}
}
- var count int
- var err os.Error
- count, err = io.ReadAtLeast(d.r, d.window[d.windowEnd:], 1)
+ count, err := d.r.Read(d.window[d.windowEnd:])
d.windowEnd += count
+ if count == 0 && err == nil {
+ d.sync = true
+ }
if err == os.EOF {
- return index, nil
+ d.eof = true
+ err = nil
}
return index, err
}
buf := make([]byte, maxStoreBlockSize)
for {
n, err := d.r.Read(buf)
- if n > 0 {
+ if n == 0 && err == nil {
+ d.sync = true
+ }
+ if n > 0 || d.sync {
if err := d.writeStoredBlock(buf[0:n]); err != nil {
return err
}
+ if d.sync {
+ d.syncChan <- nil
+ d.sync = false
+ }
}
if err != nil {
if err == os.EOF {
hash = int(d.window[index])<<hashShift + int(d.window[index+1])
}
chainHead := -1
+Loop:
for {
if index > windowEnd {
panic("index > windowEnd")
maxInsertIndex = windowEnd - (minMatchLength - 1)
lookahead = windowEnd - index
if lookahead == 0 {
- break
+ // Flush current output block if any.
+ if byteAvailable {
+ // There is still one pending token that needs to be flushed
+ tokens[ti] = literalToken(uint32(d.window[index-1]) & 0xFF)
+ ti++
+ byteAvailable = false
+ }
+ if ti > 0 {
+ if err = d.writeBlock(tokens[0:ti], index, false); err != nil {
+ return
+ }
+ ti = 0
+ }
+ if d.sync {
+ d.w.writeStoredHeader(0, false)
+ d.w.flush()
+ d.syncChan <- d.w.err
+ d.sync = false
+ }
+
+ // If this was only a sync (not at EOF) keep going.
+ if !d.eof {
+ continue
+ }
+ break Loop
}
}
if index < maxInsertIndex {
byteAvailable = true
}
}
-
- }
- if byteAvailable {
- // There is still one pending token that needs to be flushed
- tokens[ti] = literalToken(uint32(d.window[index-1]) & 0xFF)
- ti++
- }
-
- if ti > 0 {
- if err = d.writeBlock(tokens[0:ti], index, false); err != nil {
- return
- }
}
return
}
-func (d *compressor) compressor(r io.Reader, w io.Writer, level int, logWindowSize uint) (err os.Error) {
+func (d *compressor) compress(r io.Reader, w io.Writer, level int, logWindowSize uint) (err os.Error) {
d.r = r
d.w = newHuffmanBitWriter(w)
d.level = level
return WrongValueError{"level", 0, 9, int32(level)}
}
+ if d.sync {
+ d.syncChan <- err
+ d.sync = false
+ }
if err != nil {
return err
}
return d.flush()
}
-func newCompressor(w io.Writer, level int, logWindowSize uint) io.WriteCloser {
+// NewWriter returns a new Writer compressing
+// data at the given level. Following zlib, levels
+// range from 1 (BestSpeed) to 9 (BestCompression);
+// higher levels typically run slower but compress more.
+// Level 0 (NoCompression) does not attempt any
+// compression; it only adds the necessary DEFLATE framing.
+func NewWriter(w io.Writer, level int) *Writer {
+ const logWindowSize = logMaxOffsetSize
var d compressor
+ d.syncChan = make(chan os.Error, 1)
pr, pw := syncPipe()
go func() {
- err := d.compressor(pr, w, level, logWindowSize)
+ err := d.compress(pr, w, level, logWindowSize)
pr.CloseWithError(err)
}()
- return pw
+ return &Writer{pw, &d}
+}
+
+// A Writer takes data written to it and writes the compressed
+// form of that data to an underlying writer (see NewWriter).
+type Writer struct {
+ w *syncPipeWriter
+ d *compressor
+}
+
+// Write writes data to w, which will eventually write the
+// compressed form of data to its underlying writer.
+func (w *Writer) Write(data []byte) (n int, err os.Error) {
+ if len(data) == 0 {
+ // no point, and nil interferes with sync
+ return
+ }
+ return w.w.Write(data)
+}
+
+// Flush flushes any pending compressed data to the underlying writer.
+// It is useful mainly in compressed network protocols, to ensure that
+// a remote reader has enough data to reconstruct a packet.
+// Flush does not return until the data has been written.
+// If the underlying writer returns an error, Flush returns that error.
+//
+// In the terminology of the zlib library, Flush is equivalent to Z_SYNC_FLUSH.
+func (w *Writer) Flush() os.Error {
+ // For more about flushing:
+ // http://www.bolet.org/~pornin/deflate-flush.html
+ if w.d.sync {
+ panic("compress/flate: double Flush")
+ }
+ _, err := w.w.Write(nil)
+ err1 := <-w.d.syncChan
+ if err == nil {
+ err = err1
+ }
+ return err
}
-func NewWriter(w io.Writer, level int) io.WriteCloser {
- return newCompressor(w, level, logMaxOffsetSize)
+// Close flushes and closes the writer.
+func (w *Writer) Close() os.Error {
+ return w.w.Close()
}
import (
"bytes"
"fmt"
+ "io"
"io/ioutil"
"os"
+ "sync"
"testing"
)
func TestDeflate(t *testing.T) {
for _, h := range deflateTests {
- buffer := bytes.NewBuffer([]byte{})
+ buffer := bytes.NewBuffer(nil)
w := NewWriter(buffer, h.level)
w.Write(h.in)
w.Close()
}
}
+type syncBuffer struct {
+ buf bytes.Buffer
+ mu sync.RWMutex
+ closed bool
+ ready chan bool
+}
+
+func newSyncBuffer() *syncBuffer {
+ return &syncBuffer{ready: make(chan bool, 1)}
+}
+
+func (b *syncBuffer) Read(p []byte) (n int, err os.Error) {
+ for {
+ b.mu.RLock()
+ n, err = b.buf.Read(p)
+ b.mu.RUnlock()
+ if n > 0 || b.closed {
+ return
+ }
+ <-b.ready
+ }
+ panic("unreachable")
+}
+
+func (b *syncBuffer) Write(p []byte) (n int, err os.Error) {
+ n, err = b.buf.Write(p)
+ _ = b.ready <- true
+ return
+}
+
+func (b *syncBuffer) WriteMode() {
+ b.mu.Lock()
+}
+
+func (b *syncBuffer) ReadMode() {
+ b.mu.Unlock()
+ _ = b.ready <- true
+}
+
+func (b *syncBuffer) Close() os.Error {
+ b.closed = true
+ _ = b.ready <- true
+ return nil
+}
+
+func testSync(t *testing.T, level int, input []byte, name string) {
+ if len(input) == 0 {
+ return
+ }
+
+ t.Logf("--testSync %d, %d, %s", level, len(input), name)
+ buf := newSyncBuffer()
+ buf1 := new(bytes.Buffer)
+ buf.WriteMode()
+ w := NewWriter(io.MultiWriter(buf, buf1), level)
+ r := NewReader(buf)
+
+ // Write half the input and read back.
+ for i := 0; i < 2; i++ {
+ var lo, hi int
+ if i == 0 {
+ lo, hi = 0, (len(input)+1)/2
+ } else {
+ lo, hi = (len(input)+1)/2, len(input)
+ }
+ t.Logf("#%d: write %d-%d", i, lo, hi)
+ if _, err := w.Write(input[lo:hi]); err != nil {
+ t.Errorf("testSync: write: %v", err)
+ return
+ }
+ if i == 0 {
+ if err := w.Flush(); err != nil {
+ t.Errorf("testSync: flush: %v", err)
+ return
+ }
+ } else {
+ if err := w.Close(); err != nil {
+ t.Errorf("testSync: close: %v", err)
+ }
+ }
+ buf.ReadMode()
+ out := make([]byte, hi-lo+1)
+ m, err := io.ReadAtLeast(r, out, hi-lo)
+ t.Logf("#%d: read %d", i, m)
+ if m != hi-lo || err != nil {
+ t.Errorf("testSync/%d (%d, %d, %s): read %d: %d, %v (%d left)", i, level, len(input), name, hi-lo, m, err, buf.buf.Len())
+ return
+ }
+ if !bytes.Equal(input[lo:hi], out[:hi-lo]) {
+ t.Errorf("testSync/%d: read wrong bytes: %x vs %x", i, input[lo:hi], out[:hi-lo])
+ return
+ }
+ if i == 0 && buf.buf.Len() != 0 {
+ t.Errorf("testSync/%d (%d, %d, %s): extra data after %d", i, level, len(input), name, hi-lo)
+ }
+ buf.WriteMode()
+ }
+ buf.ReadMode()
+ out := make([]byte, 10)
+ if n, err := r.Read(out); n > 0 || err != os.EOF {
+ t.Errorf("testSync (%d, %d, %s): final Read: %d, %v (hex: %x)", level, len(input), name, n, err, out[0:n])
+ }
+ if buf.buf.Len() != 0 {
+ t.Errorf("testSync (%d, %d, %s): extra data at end", level, len(input), name)
+ }
+ r.Close()
+
+ // stream should work for ordinary reader too
+ r = NewReader(buf1)
+ out, err := ioutil.ReadAll(r)
+ if err != nil {
+ t.Errorf("testSync: read: %s", err)
+ return
+ }
+ r.Close()
+ if !bytes.Equal(input, out) {
+ t.Errorf("testSync: decompress(compress(data)) != data: level=%d input=%s", level, name)
+ }
+}
+
+
func testToFromWithLevel(t *testing.T, level int, input []byte, name string) os.Error {
- buffer := bytes.NewBuffer([]byte{})
+ buffer := bytes.NewBuffer(nil)
w := NewWriter(buffer, level)
w.Write(input)
w.Close()
- decompressor := NewReader(buffer)
- decompressed, err := ioutil.ReadAll(decompressor)
+ r := NewReader(buffer)
+ out, err := ioutil.ReadAll(r)
if err != nil {
- t.Errorf("reading decompressor: %s", err)
+ t.Errorf("read: %s", err)
return err
}
- decompressor.Close()
- if bytes.Compare(input, decompressed) != 0 {
+ r.Close()
+ if !bytes.Equal(input, out) {
t.Errorf("decompress(compress(data)) != data: level=%d input=%s", level, name)
}
+
+ testSync(t, level, input, name)
return nil
}
// Output history, buffer.
hist [maxHist]byte
hp int // current output position in buffer
+ hw int // have written hist[0:hw] already
hfull bool // buffer has filled at least once
// Temporary buffer (avoids repeated allocation).
return CorruptInputError(f.roffset)
}
+ if n == 0 {
+ // 0-length block means sync
+ return f.flush()
+ }
+
// Read len bytes into history,
// writing as history fills.
for n > 0 {
// Flush any buffered output to the underlying writer.
func (f *decompressor) flush() os.Error {
- if f.hp == 0 {
+ if f.hw == f.hp {
return nil
}
- n, err := f.w.Write(f.hist[0:f.hp])
- if n != f.hp && err == nil {
+ n, err := f.w.Write(f.hist[f.hw:f.hp])
+ if n != f.hp-f.hw && err == nil {
err = io.ErrShortWrite
}
if err != nil {
return &WriteError{f.woffset, err}
}
- f.woffset += int64(f.hp)
- f.hp = 0
- f.hfull = true
+ f.woffset += int64(f.hp - f.hw)
+ f.hw = f.hp
+ if f.hp == len(f.hist) {
+ f.hp = 0
+ f.hw = 0
+ f.hfull = true
+ }
return nil
}
return bufio.NewReader(r)
}
-// Inflate reads DEFLATE-compressed data from r and writes
+// decompress reads DEFLATE-compressed data from r and writes
// the uncompressed data to w.
-func (f *decompressor) decompressor(r io.Reader, w io.Writer) os.Error {
+func (f *decompressor) decompress(r io.Reader, w io.Writer) os.Error {
f.r = makeReader(r)
f.w = w
f.woffset = 0
func NewReader(r io.Reader) io.ReadCloser {
var f decompressor
pr, pw := io.Pipe()
- go func() { pw.CloseWithError(f.decompressor(r, pw)) }()
+ go func() { pw.CloseWithError(f.decompress(r, pw)) }()
return pr
}