From 03739d2b113afe60638069c4e1604dc2ac27380d Mon Sep 17 00:00:00 2001
From: Eugene Kliuchnikov <eustas@google.com>
Date: Mon, 29 May 2017 17:55:14 +0200
Subject: Update (#555)

Update:
 * new CLI; bro -> brotli; + man page
 * JNI wrappers preparation (for bazel build)
 * add raw binary dictionary representation `dictionary.bin`
 * add ability to side-load brotli RFC dictionary
 * decoder persists last error now
 * fix `BrotliDecoderDecompress` documentation
 * go reader don't block until necessary
 * more consistent bazel target names
 * Java dictionary data compiled footprint reduced
 * Java tests refactoring
---
 go/cbrotli/cbrotli_test.go | 81 ++++++++++++++++++++++++++++++++++++++++++++++
 go/cbrotli/reader.go       |  9 ++++--
 2 files changed, 88 insertions(+), 2 deletions(-)

(limited to 'go')

diff --git a/go/cbrotli/cbrotli_test.go b/go/cbrotli/cbrotli_test.go
index 0a9368b..f59ec58 100755
--- a/go/cbrotli/cbrotli_test.go
+++ b/go/cbrotli/cbrotli_test.go
@@ -9,9 +9,11 @@ import (
 	"bytes"
 	"fmt"
 	"io"
+	"io/ioutil"
 	"math"
 	"math/rand"
 	"testing"
+	"time"
 )
 
 func checkCompressedData(compressedData, wantOriginalData []byte) error {
@@ -173,6 +175,85 @@ func TestEncoderFlush(t *testing.T) {
 	}
 }
 
+type readerWithTimeout struct {
+	io.ReadCloser
+}
+
+func (r readerWithTimeout) Read(p []byte) (int, error) {
+	type result struct {
+		n   int
+		err error
+	}
+	ch := make(chan result)
+	go func() {
+		n, err := r.ReadCloser.Read(p)
+		ch <- result{n, err}
+	}()
+	select {
+	case result := <-ch:
+		return result.n, result.err
+	case <-time.After(5 * time.Second):
+		return 0, fmt.Errorf("read timed out")
+	}
+}
+
+func TestDecoderStreaming(t *testing.T) {
+	pr, pw := io.Pipe()
+	writer := NewWriter(pw, WriterOptions{Quality: 5, LGWin: 20})
+	reader := readerWithTimeout{NewReader(pr)}
+	defer func() {
+		if err := reader.Close(); err != nil {
+			t.Errorf("reader.Close: %v", err)
+		}
+		go ioutil.ReadAll(pr) // swallow the "EOF" token from writer.Close
+		if err := writer.Close(); err != nil {
+			t.Errorf("writer.Close: %v", err)
+		}
+	}()
+
+	ch := make(chan []byte)
+	errch := make(chan error)
+	go func() {
+		for {
+			segment, ok := <-ch
+			if !ok {
+				return
+			}
+			if n, err := writer.Write(segment); err != nil || n != len(segment) {
+				errch <- fmt.Errorf("write=%v,%v, want %v,%v", n, err, len(segment), nil)
+				return
+			}
+			if err := writer.Flush(); err != nil {
+				errch <- fmt.Errorf("flush: %v", err)
+				return
+			}
+		}
+	}()
+	defer close(ch)
+
+	segments := [...][]byte{
+		[]byte("first"),
+		[]byte("second"),
+		[]byte("third"),
+	}
+	for k, segment := range segments {
+		t.Run(fmt.Sprintf("Segment%d", k), func(t *testing.T) {
+			select {
+			case ch <- segment:
+			case err := <-errch:
+				t.Fatalf("write: %v", err)
+			case <-time.After(5 * time.Second):
+				t.Fatalf("timed out")
+			}
+			wantLen := len(segment)
+			got := make([]byte, wantLen)
+			if n, err := reader.Read(got); err != nil || n != wantLen || !bytes.Equal(got, segment) {
+				t.Fatalf("read[%d]=%q,%v,%v, want %q,%v,%v", k, got, n, err, segment, wantLen, nil)
+			}
+		})
+	}
+}
+
 func TestReader(t *testing.T) {
 	content := bytes.Repeat([]byte("hello world!"), 10000)
 	encoded, _ := Encode(content, WriterOptions{Quality: 5})
diff --git a/go/cbrotli/reader.go b/go/cbrotli/reader.go
index a05809b..3d8d424 100755
--- a/go/cbrotli/reader.go
+++ b/go/cbrotli/reader.go
@@ -95,7 +95,7 @@ func (r *Reader) Read(p []byte) (n int, err error) {
 		return 0, nil
 	}
 
-	for n == 0 {
+	for {
 		var written, consumed C.size_t
 		var data *C.uint8_t
 		if len(r.in) != 0 {
@@ -128,9 +128,14 @@ func (r *Reader) Read(p []byte) (n int, err error) {
 			return 0, errInvalidState
 		}
 
+		// Calling r.src.Read may block. Don't block if we have data to return.
+		if n > 0 {
+			return n, nil
+		}
+
 		// Top off the buffer.
 		encN, err := r.src.Read(r.buf)
-		if encN == 0 && n == 0 {
+		if encN == 0 {
 			// Not enough data to complete decoding.
 			if err == io.EOF {
 				return 0, io.ErrUnexpectedEOF
-- 
cgit v1.1