Enable calling mirrorPool reader Read after EOF

pkg/filter/proxy/masterslavereader.go

@@ -23,40 +23,45 @@ import (
 )
 
 type (
-	// masterSlaveReader reads bytes to master,
+	// primarySecondaryReader reads bytes to master,
 	// and synchronize them to slave.
 	// Currently only support one slave.
-	masterSlaveReader struct {
-		masterReader io.Reader
-		slaveReader  io.Reader
+	primarySecondaryReader struct {
+		primaryReader io.Reader
+		secondaryReader  io.Reader
 	}
 
-	masterReader struct {
+	primaryReader struct {
 		r        io.Reader
 		buffChan chan []byte
+		sawEOF   bool
 	}
 
-	slaveReader struct {
+	secondaryReader struct {
 		unreadBuff *bytes.Buffer
 		buffChan   chan []byte
 	}
 )
 
-func newMasterSlaveReader(r io.Reader) (io.ReadCloser, io.Reader) {
+func newPrimarySecondaryReader(r io.Reader) (io.ReadCloser, io.Reader) {
 	buffChan := make(chan []byte, 10)
-	mr := &masterReader{
+	mr := &primaryReader{
 		r:        r,
 		buffChan: buffChan,
+		sawEOF:   false,
 	}
-	sr := &slaveReader{
+	sr := &secondaryReader{
 		unreadBuff: bytes.NewBuffer(nil),
 		buffChan:   buffChan,
 	}
 
 	return mr, sr
 }
 
-func (mr *masterReader) Read(p []byte) (n int, err error) {
+func (mr *primaryReader) Read(p []byte) (n int, err error) {
+	if mr.sawEOF {
+		return 0, io.EOF
+	}
 	buff := bytes.NewBuffer(nil)
 	tee := io.TeeReader(mr.r, buff)
 	n, err = tee.Read(p)
@@ -67,20 +72,21 @@ func (mr *masterReader) Read(p []byte) (n int, err error) {
 
 	if err == io.EOF {
 		close(mr.buffChan)
+		mr.sawEOF = true
 	}
 
 	return n, err
 }
 
-func (mr *masterReader) Close() error {
+func (mr *primaryReader) Close() error {
 	if closer, ok := mr.r.(io.ReadCloser); ok {
 		return closer.Close()
 	}
 
 	return nil
 }
 
-func (sr *slaveReader) Read(p []byte) (int, error) {
+func (sr *secondaryReader) Read(p []byte) (int, error) {
 	buff, ok := <-sr.buffChan
 
 	if !ok {

pkg/filter/proxy/masterslavereader_test.go

@@ -31,7 +31,7 @@ func TestReader(t *testing.T) {
 
 	r := bytes.NewReader(b.Bytes())
 
-	reader1, reader2 := newMasterSlaveReader(r)
+	reader1, reader2 := newPrimarySecondaryReader(r)
 
 	buff := make([]byte, 10)
 	n, _ := reader1.Read(buff)
@@ -56,7 +56,7 @@ func TestReader(t *testing.T) {
 func TestEOFReader(t *testing.T) {
 	eofReader := iotest.ErrReader(io.EOF)
 
-	reader1, reader2 := newMasterSlaveReader(eofReader)
+	reader1, reader2 := newPrimarySecondaryReader(eofReader)
 
 	buff := make([]byte, 10)
 	_, err := reader1.Read(buff)
@@ -77,3 +77,12 @@ func TestEOFReader(t *testing.T) {
 
 	reader1.Close()
 }
+
+func TestMultipleRead(t *testing.T) {
+	b := bytes.NewReader([]byte("abc"))
+	reader1, _ := newPrimarySecondaryReader(b)
+
+	reader1.Read(make([]byte, 10))
+	reader1.Read(make([]byte, 10))
+	reader1.Read(make([]byte, 10))
+}

pkg/filter/proxy/proxy.go

@@ -317,7 +317,7 @@ func (b *Proxy) Handle(ctx context.HTTPContext) (result string) {
 
 func (b *Proxy) handle(ctx context.HTTPContext) (result string) {
 	if b.mirrorPool != nil && b.mirrorPool.filter.Filter(ctx) {
-		master, slave := newMasterSlaveReader(ctx.Request().Body())
+		master, slave := newPrimarySecondaryReader(ctx.Request().Body())
 		ctx.Request().SetBody(master)
 
 		wg := &sync.WaitGroup{}