Merge pull request nsqio#996 from andyxning/fix_memory_leak_with_large_message

nsqd: fix memory leak with large message

Author: mreiferson

nsqd/protocol_v2.go

@@ -120,10 +120,10 @@ func (p *protocolV2) IOLoop(conn net.Conn) error {
 	return err
 }
 
-func (p *protocolV2) SendMessage(client *clientV2, msg *Message, buf *bytes.Buffer) error {
+func (p *protocolV2) SendMessage(client *clientV2, msg *Message) error {
 	p.ctx.nsqd.logf(LOG_DEBUG, "PROTOCOL(V2): writing msg(%s) to client(%s) - %s", msg.ID, client, msg.Body)
+	var buf = &bytes.Buffer{}
 
-	buf.Reset()
 	_, err := msg.WriteTo(buf)
 	if err != nil {
 		return err
@@ -199,7 +199,6 @@ func (p *protocolV2) Exec(client *clientV2, params [][]byte) ([]byte, error) {
 
 func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
 	var err error
-	var buf bytes.Buffer
 	var memoryMsgChan chan *Message
 	var backendMsgChan chan []byte
 	var subChannel *Channel
@@ -312,7 +311,7 @@ func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
 
 			subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
 			client.SendingMessage()
-			err = p.SendMessage(client, msg, &buf)
+			err = p.SendMessage(client, msg)
 			if err != nil {
 				goto exit
 			}
@@ -325,7 +324,7 @@ func (p *protocolV2) messagePump(client *clientV2, startedChan chan bool) {
 
 			subChannel.StartInFlightTimeout(msg, client.ID, msgTimeout)
 			client.SendingMessage()
-			err = p.SendMessage(client, msg, &buf)
+			err = p.SendMessage(client, msg)
 			if err != nil {
 				goto exit
 			}