Merge pull request #39 from felixhao/fix_reset

reset sub only in need

CHANGELOG.md

@@ -1,4 +1,6 @@
 # Overlord
+## Version 1.5.1
+1. reset sub message only in nedd.
 
 ## Version 1.5.0
 1. refactor message pipeline.

proto/memcache/pinger.go

@@ -12,7 +12,7 @@ import (
 )
 
 const (
-	pingBufferSize = 8
+	pingBufferSize = 128
 )
 
 var (
@@ -32,8 +32,8 @@ type mcPinger struct {
 func NewPinger(nc *libnet.Conn) proto.Pinger {
 	return &mcPinger{
 		conn: nc,
-		bw:   bufio.NewWriter(nc),
 		br:   bufio.NewReader(nc, bufio.NewBuffer(pingBufferSize)),
+		bw:   bufio.NewWriter(nc),
 	}
 }
 

proto/message.go

@@ -90,6 +90,7 @@ func (m *Message) Reset() {
 // clear will clean the msg
 func (m *Message) clear() {
 	m.Reset()
+	m.reqn = 0
 	m.req = nil
 	m.wg = nil
 	m.subs = nil
@@ -127,7 +128,10 @@ func (m *Message) MarkEnd() {
 
 // ResetSubs will return the Msg data to flush and reset
 func (m *Message) ResetSubs() {
-	for i := range m.subs {
+	if !m.IsBatch() {
+		return
+	}
+	for i := range m.subs[:m.reqn] {
 		m.subs[i].Reset()
 	}
 	m.reqn = 0
@@ -226,7 +230,10 @@ func (m *Message) Err() error {
 	if m.err != nil {
 		return m.err
 	}
-	for _, s := range m.subs {
+	if !m.IsBatch() {
+		return nil
+	}
+	for _, s := range m.subs[:m.reqn] {
 		if s.err != nil {
 			return s.err
 		}

proto/message_test.go

@@ -0,0 +1,64 @@
+package proto
+
+import (
+	"errors"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestMessage(t *testing.T) {
+	msgs := GetMsgs(1)
+	assert.Len(t, msgs, 1)
+	PutMsgs(msgs)
+	msgs = GetMsgs(1, 1)
+	assert.Len(t, msgs, 1)
+	PutMsgs(msgs)
+
+	msg := NewMessage()
+	msg.Reset()
+	msg.clear()
+	msg.WithRequest(&mockRequest{})
+	msg.ResetSubs()
+
+	msg.WithRequest(&mockRequest{})
+	req := msg.Request()
+	assert.NotNil(t, req)
+	reqs := msg.Requests()
+	assert.Len(t, reqs, 2)
+	isb := msg.IsBatch()
+	assert.True(t, isb)
+	msgs = msg.Batch()
+	assert.Len(t, msgs, 2)
+
+	msg.ResetSubs()
+	req = msg.NextReq()
+	assert.NotNil(t, req)
+
+	wg := &sync.WaitGroup{}
+	msg.WithWaitGroup(wg)
+	msg.Add()
+	msg.Done()
+
+	msg.MarkStart()
+	time.Sleep(time.Millisecond * 50)
+	msg.MarkRead()
+	time.Sleep(time.Millisecond * 50)
+	msg.MarkWrite()
+	time.Sleep(time.Millisecond * 50)
+	msg.MarkEnd()
+	ts := msg.TotalDur()
+	assert.NotZero(t, ts)
+	ts = msg.RemoteDur()
+	assert.NotZero(t, ts)
+
+	msg.WithError(errors.New("some error"))
+	err := msg.Err()
+	assert.EqualError(t, err, "some error")
+
+	emsg := ErrMessage(errors.New("some error"))
+	err = emsg.Err()
+	assert.EqualError(t, err, "some error")
+}

proto/redis/pinger.go

@@ -12,6 +12,10 @@ import (
 	"github.com/pkg/errors"
 )
 
+const (
+	pingBufferSize = 128
+)
+
 // errors
 var (
 	ErrPingClosed = errs.New("ping interface has been closed")
@@ -36,7 +40,7 @@ type pinger struct {
 func NewPinger(conn *libnet.Conn) proto.Pinger {
 	return &pinger{
 		conn:  conn,
-		br:    bufio.NewReader(conn, bufio.NewBuffer(7)),
+		br:    bufio.NewReader(conn, bufio.NewBuffer(pingBufferSize)),
 		bw:    bufio.NewWriter(conn),
 		state: opened,
 	}

proto/redis/pinger_test.go

@@ -3,8 +3,6 @@ package redis
 import (
 	"testing"
 
-	"overlord/lib/bufio"
-
 	"github.com/pkg/errors"
 	"github.com/stretchr/testify/assert"
 )
@@ -29,7 +27,7 @@ func TestPingerWrongResp(t *testing.T) {
 	conn := _createConn([]byte("-Error: iam more than 7 bytes\r\n"))
 	p := NewPinger(conn)
 	err := p.Ping()
-	assert.Equal(t, bufio.ErrBufferFull, errors.Cause(err))
+	assert.Equal(t, ErrBadPong, errors.Cause(err))
 
 	conn = _createConn([]byte("-Err\r\n"))
 	p = NewPinger(conn)