streamWriter

Author: Michael Lauer

protocol.go

@@ -113,9 +113,11 @@ func readRecord(r io.Reader) (record, error) {
 	if _, err := r.Read(pad[:1]); err != nil {
 		return rec, err
 	}
-	rec.Content = make([]byte, clength)
-	if _, err := r.Read(rec.Content); err != nil {
-		return rec, err
+	if clength != 0 {
+		rec.Content = make([]byte, clength)
+		if _, err := r.Read(rec.Content); err != nil {
+			return rec, err
+		}
 	}
 	if plength != 0 {
 		if _, err := r.Read(pad[:plength]); err != nil {

protocol_test.go

@@ -28,6 +28,14 @@ func TestWriteRecord(t *testing.T) {
 			},
 			[]byte{1, 3, 0, 1, 0, 8, 0, 0},
 		},
+		{
+			record{
+				Type:    3,
+				Id:      1,
+				Content: nil,
+			},
+			[]byte{1, 3, 0, 1, 0, 0, 0, 0},
+		},
 	}
 	for _, d := range data {
 		rec := d.rec

streams.go

@@ -0,0 +1,52 @@
+package gofcgisrv
+
+import (
+	"bytes"
+	"io"
+	"sync"
+)
+
+type streamReader struct {
+	buffer  bytes.Buffer
+	lock    sync.Mutex
+	gotData *sync.Cond
+	err     error
+}
+
+func newStreamReader() *streamReader {
+	s := new(streamReader)
+	s.gotData = sync.NewCond(&s.lock)
+	return s
+}
+
+func (sr *streamReader) Read(data []byte) (int, error) {
+	sr.lock.Lock()
+	defer sr.lock.Unlock()
+	// Wait for something to show up
+	for sr.buffer.Len() == 0 && sr.err == nil {
+		sr.gotData.Wait()
+	}
+	if sr.buffer.Len() == 0 {
+		return 0, sr.err
+	}
+	return sr.buffer.Read(data)
+}
+
+func (sr *streamReader) Write(data []byte) (int, error) {
+	sr.lock.Lock()
+	defer sr.lock.Unlock()
+	if sr.err == nil {
+		n, err := sr.buffer.Write(data)
+		sr.gotData.Signal()
+		return n, err
+	}
+	return 0, sr.err
+}
+
+func (sr *streamReader) Close() error {
+	sr.lock.Lock()
+	defer sr.lock.Unlock()
+	sr.err = io.EOF
+	sr.gotData.Signal()
+	return nil
+}

streams_test.go

@@ -0,0 +1,51 @@
+package gofcgisrv
+
+import (
+	"io"
+	"io/ioutil"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestStreamReader(t *testing.T) {
+	sr := newStreamReader()
+	io.WriteString(sr, "abc")
+	data := make([]byte, 1024)
+	n, err := sr.Read(data)
+	if err != nil {
+		t.Errorf("Read error %v", err)
+	}
+	if string(data[:n]) != "abc" {
+		t.Errorf("Read %s, not %s", data[:n], "abc")
+	}
+
+	ch := make(chan string)
+	go func() {
+		n, err := sr.Read(data)
+		if err != nil {
+			t.Error(err)
+		}
+		ch <- string(data[:n])
+	}()
+	// Let the goroutine execute a bit
+	time.Sleep(time.Millisecond)
+	go func() {
+		sr.Write([]byte("ABCD"))
+	}()
+	str := <-ch
+	if str != "ABCD" {
+		t.Errorf("Read %s, not %s", str, "ABCD")
+	}
+
+	ss := []string{"foo", "bar", strings.Repeat("xyzå", 100)}
+	for _, s := range ss {
+		sr.Write([]byte(s))
+	}
+	go sr.Close()
+	ssj := strings.Join(ss, "")
+	read, err := ioutil.ReadAll(sr)
+	if err != nil || string(read) != ssj {
+		t.Errorf("Read %s with error %v", read, err)
+	}
+}