Merge pull request #506 from go-redis/fix/separate-buffer

Separate read and write buffers for PubSub.

internal/pool/conn.go

@@ -21,12 +21,11 @@ type Conn struct {
 }
 
 func NewConn(netConn net.Conn) *Conn {
-	buf := make([]byte, 4096)
 	cn := &Conn{
 		netConn: netConn,
-		Wb:      proto.NewWriteBuffer(buf),
+		Wb:      proto.NewWriteBuffer(),
 	}
-	cn.Rd = proto.NewReader(cn.netConn, buf)
+	cn.Rd = proto.NewReader(cn.netConn)
 	cn.SetUsedAt(time.Now())
 	return cn
 }

internal/proto/reader.go

@@ -26,10 +26,10 @@ type Reader struct {
 	buf []byte
 }
 
-func NewReader(rd io.Reader, buf []byte) *Reader {
+func NewReader(rd io.Reader) *Reader {
 	return &Reader{
 		src: bufio.NewReader(rd),
-		buf: buf,
+		buf: make([]byte, 4096),
 	}
 }
 

internal/proto/reader_test.go

@@ -14,18 +14,18 @@ import (
 var _ = Describe("Reader", func() {
 
 	It("should read n bytes", func() {
-		data, err := proto.NewReader(strings.NewReader("ABCDEFGHIJKLMNO"), nil).ReadN(10)
+		data, err := proto.NewReader(strings.NewReader("ABCDEFGHIJKLMNO")).ReadN(10)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(len(data)).To(Equal(10))
 		Expect(string(data)).To(Equal("ABCDEFGHIJ"))
 
-		data, err = proto.NewReader(strings.NewReader(strings.Repeat("x", 8192)), nil).ReadN(6000)
+		data, err = proto.NewReader(strings.NewReader(strings.Repeat("x", 8192))).ReadN(6000)
 		Expect(err).NotTo(HaveOccurred())
 		Expect(len(data)).To(Equal(6000))
 	})
 
 	It("should read lines", func() {
-		p := proto.NewReader(strings.NewReader("$5\r\nhello\r\n"), nil)
+		p := proto.NewReader(strings.NewReader("$5\r\nhello\r\n"))
 
 		data, err := p.ReadLine()
 		Expect(err).NotTo(HaveOccurred())
@@ -63,7 +63,7 @@ func benchmarkParseReply(b *testing.B, reply string, m proto.MultiBulkParse, wan
 	for i := 0; i < b.N; i++ {
 		buf.WriteString(reply)
 	}
-	p := proto.NewReader(buf, nil)
+	p := proto.NewReader(buf)
 	b.ResetTimer()
 
 	for i := 0; i < b.N; i++ {

internal/proto/write_buffer.go

@@ -12,9 +12,9 @@ type WriteBuffer struct {
 	b []byte
 }
 
-func NewWriteBuffer(b []byte) *WriteBuffer {
+func NewWriteBuffer() *WriteBuffer {
 	return &WriteBuffer{
-		b: b,
+		b: make([]byte, 0, 4096),
 	}
 }
 

internal/proto/write_buffer_test.go

@@ -14,7 +14,7 @@ var _ = Describe("WriteBuffer", func() {
 	var buf *proto.WriteBuffer
 
 	BeforeEach(func() {
-		buf = proto.NewWriteBuffer(nil)
+		buf = proto.NewWriteBuffer()
 	})
 
 	It("should reset", func() {
@@ -53,7 +53,7 @@ var _ = Describe("WriteBuffer", func() {
 })
 
 func BenchmarkWriteBuffer_Append(b *testing.B) {
-	buf := proto.NewWriteBuffer(nil)
+	buf := proto.NewWriteBuffer()
 	args := []interface{}{"hello", "world", "foo", "bar"}
 
 	for i := 0; i < b.N; i++ {