ronantakizawa/codereview-bench huggingface dataset:
Repository: nsqio/nsq
File: nsqd/channel.go (Go)

Code before:
}

// BackendQueue implements the Queue interface
func (c *Channel) BackendQueue() BackendQueue {
	return c.backend
}

// InFlight implements the Queue interface
func (c *Channel) InFlight() map[string]*pqueue.Item {
	return c.inFlightMessages
}

// Deferred implements the Queue interface
func (c *Channel) Deferred() map[string]*pqueue.Item {
	return c.deferredMessages
}

func (c *Channel) Depth() int64 {
	return int64(len(c.memoryMsgChan)) + c.backend.Depth() + int64(atomic.LoadInt32(&c.bufferedCount))
}

// PutMessage writes to the appropriate incoming message channel
// (which will be routed asynchronously)
func (c *Channel) PutMessage(msg *nsq.Message) error {
	c.RLock()
	defer c.RUnlock()
	if atomic.LoadInt32(&c.exitFlag) == 1 {
		return errors.New("exiting")
	}
	c.incomingMsgChan <- msg
	atomic.AddUint64(&c.messageCount, 1)
	return nil
}

// FinishMessage successfully discards an in-flight message
func (c *Channel) FinishMessage(client ClientStatTracker, id []byte) error {
	item, err := c.popInFlightMessage(client, id)
	if err != nil {
		log.Printf("ERROR: failed to finish message(%s) - %s", id, err.Error())
	} else {
		c.removeFromInFlightPQ(item)
	}
	return err
}

// RequeueMessage requeues a message based on `time.Duration`, ie:
//
// `timeoutMs` == 0 - requeue a message immediately
// `timeoutMs`  > 0 - asynchronously wait for the specified timeout
//     and requeue a message (aka "deferred requeue")
//

Review comment (suggestion):
its applied to the channel, so any new clients pick it up automatically...

the loop below (if you look at the implementation `client_v2.Pause()`) just pushes a state update message


Code after:
// MemoryChan implements the Queue interface
func (c *Channel) MemoryChan() chan *nsq.Message {
	return c.memoryMsgChan
}

// BackendQueue implements the Queue interface
func (c *Channel) BackendQueue() BackendQueue {
	return c.backend
}

// InFlight implements the Queue interface
func (c *Channel) InFlight() map[string]*pqueue.Item {
	return c.inFlightMessages
}

// Deferred implements the Queue interface
func (c *Channel) Deferred() map[string]*pqueue.Item {
	return c.deferredMessages
}

func (c *Channel) Depth() int64 {
	return int64(len(c.memoryMsgChan)) + c.backend.Depth() + int64(atomic.LoadInt32(&c.bufferedCount))
}

func (c *Channel) Pause() {
	atomic.StoreInt32(&c.paused, 1)
	c.RLock()
	defer c.RUnlock()
	for _, client := range c.clients {
		client.Pause()
	}
}

func (c *Channel) UnPause() {
	atomic.StoreInt32(&c.paused, 0)
	c.RLock()
	defer c.RUnlock()
	for _, client := range c.clients {
		client.UnPause()
	}
}

func (c *Channel) IsPaused() bool {
	return atomic.LoadInt32(&c.paused) == 1
}

// PutMessage writes to the appropriate incoming message channel
// (which will be routed asynchronously)
func (c *Channel) PutMessage(msg *nsq.Message) error {
	c.RLock()
	defer c.RUnlock()
Repository: nsqio/nsq
File: nsqd/channel_test.go (Go)

Code before:
package main

import (
	"../nsq"
	"github.com/bmizerany/assert"
	"io/ioutil"
	"log"
	"os"
	"strconv"
	"testing"
	"time"
)

// ensure that we can push a message through a topic and get it out of a channel
func TestPutMessage(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	nsqd := NewNSQd(1, NewNsqdOptions())
	defer nsqd.Exit()

	topicName := "test_put_message" + strconv.Itoa(int(time.Now().Unix()))
	topic := nsqd.GetTopic(topicName)
	channel1 := topic.GetChannel("ch")

	var id nsq.MessageID
	msg := nsq.NewMessage(id, []byte("test"))
	topic.PutMessage(msg)

	outputMsg := <-channel1.clientMsgChan
	assert.Equal(t, msg.Id, outputMsg.Id)
	assert.Equal(t, msg.Body, outputMsg.Body)
}

// ensure that both channels get the same message
func TestPutMessage2Chan(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	nsqd := NewNSQd(1, NewNsqdOptions())
	defer nsqd.Exit()

	topicName := "test_put_message_2chan" + strconv.Itoa(int(time.Now().Unix()))
	topic := nsqd.GetTopic(topicName)

Review comment (suggestion):
I see the commit comment as to why you made this change.

I've struggled with the best way to pass around instance state and admittedly things still aren't perfect.

Going to think about this a bit...


Code after:
package main

import (
	"../nsq"
	"github.com/bmizerany/assert"
	"io/ioutil"
	"log"
	"os"
	"strconv"
	"testing"
	"time"
)

// ensure that we can push a message through a topic and get it out of a channel
func TestPutMessage(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	nsqd = NewNSQd(1, NewNsqdOptions())
	defer nsqd.Exit()

	topicName := "test_put_message" + strconv.Itoa(int(time.Now().Unix()))
	topic := nsqd.GetTopic(topicName)
	channel1 := topic.GetChannel("ch")

	var id nsq.MessageID
	msg := nsq.NewMessage(id, []byte("test"))
	topic.PutMessage(msg)

	outputMsg := <-channel1.clientMsgChan
	assert.Equal(t, msg.Id, outputMsg.Id)
	assert.Equal(t, msg.Body, outputMsg.Body)
}

// ensure that both channels get the same message
func TestPutMessage2Chan(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	nsqd = NewNSQd(1, NewNsqdOptions())
	defer nsqd.Exit()

	topicName := "test_put_message_2chan" + strconv.Itoa(int(time.Now().Unix()))
	topic := nsqd.GetTopic(topicName)
Repository: nsqio/nsq
File: nsq/writer.go (Go)

Code before:
	case error:
		cmd = Nop()
	default:
		return errors.New("Body not acceptable")
	}
	msg := WMessage{
		Cmd:  cmd,
		Stat: make(chan error),
	}
	this.MessageChan <- msg
	err = <-msg.Stat
	return
}

// always retry to connect nsqd
func (this *Writer) ConnectToNSQ(addr string) error {
	var err error
	this.Conn, err = net.DialTimeout("tcp", addr, time.Second*5)
	if err != nil {
		return errors.New("fail to connect nsq")
	}
	this.Conn.SetWriteDeadline(time.Now().Add(this.WriteTimeout))
	this.Conn.Write(MagicV2)
	if err := this.identify(); err != nil {
		this.Conn.Close()
		return err
	}
	go this.writeloop()
	go this.readloop()
	go this.retry_connect(addr)
	return nil
}

func (this *Writer) identify() error {
	ci := make(map[string]interface{})
	ci["short_id"] = this.ShortIdentifier
	ci["long_id"] = this.LongIdentifier
	ci["heartbeat_interval"] = this.HeartBeatInterval
	cmd, err := Identify(ci)
	if err = cmd.Write(this.Conn); err != nil {
		log.Println("write identify failed", err)
		return err
	}
	rbuf := bufio.NewReader(this.Conn)
	resp, err := ReadResponse(rbuf)
	if err != nil {
		return err
	}
	_, data, err := UnpackResponse(resp)
	if err != nil {
		return errors.New("unpack failed")

Review comment (suggestion):
should this check to see if we've already been called so that we dont try to force a connection and launch a new set of goroutines?

you can use [`sync/atomic`](http://golang.org/pkg/sync/atomic/)


Code after:
	return this.sendCommand(cmd)
}

func (this *Writer) MultiPublish(topic string, body [][]byte) (int32, []byte, error) {
	cmd, err := MultiPublish(topic, body)
	if err != nil {
		return -1, nil, err
	}
	return this.sendCommand(cmd)
}

func (this *Writer) sendCommand(cmd *Command) (int32, []byte, error) {
	if atomic.LoadInt32(&this.state) != StateConnected {
		return -1, nil, errors.New("not connected")
	}
	t := &writerTransaction{
		cmd:       cmd,
		frameType: -1,
		doneChan:  make(chan int),
	}
	this.transactionChan <- t
	<-t.doneChan
	return t.frameType, t.data, t.err
}

func (this *Writer) ConnectToNSQ(addr string) error {
	if atomic.LoadInt32(&this.stopFlag) == 1 {
		return errors.New("writer stopped")
	}
	this.Lock()
	defer this.Unlock()
	if !atomic.CompareAndSwapInt32(&this.state,
		StateDisconnected, StateConnected) {
		return nil
	}
	var err error
	this.Conn, err = net.DialTimeout("tcp", addr, time.Second*5)
	if err != nil {
		return errors.New("failed to connect nsq")
	}
	this.SetWriteDeadline(time.Now().Add(this.WriteTimeout))
	if _, err := this.Conn.Write(MagicV2); err != nil {
		this.Close()
		return err
	}
	atomic.StoreInt32(&this.state, StateConnected)
	ci := make(map[string]interface{})
	ci["short_id"] = this.ShortIdentifier
	ci["long_id"] = this.LongIdentifier
	ci["heartbeat_interval"] = this.HeartbeatInterval
	ci["feature_negotiation"] = true
Repository: nsqio/nsq
File: nsq/writer.go (Go)

Code before:
	}
}

//send heartbeat
func (this *Writer) heartbeat() error {
	this.Conn.SetWriteDeadline(time.Now().Add(this.WriteTimeout))
	if err := Nop().Write(this.Conn); err != nil {
		return err
	}
	return nil
}

// cleanup transactions
func (this *Writer) transactionCleanup() {
	this.Lock()
	defer this.Lock()
	for _, t := range this.transactions {
		t.err = errors.New("Not connected")
		t.doneChan <- 1
	}
	this.transactions = this.transactions[:0]
}
func (this *Writer) readloop() {
	rbuf := bufio.NewReader(this.Conn)
	for {
		resp, err := ReadResponse(rbuf)
		if err != nil {
			this.Conn.Close()
			break
		}
		select {
		case this.dataChan <- resp:
		case <-this.exitChan:
			return
		}
	}
}


Review comment (bug):
we should also log something here, in the format:

``` go
log.Printf("ERROR: failed to send heartbeat to %s - %s", this.Conn, err.Error())
```


Code after:
	return err
}

func (this *Writer) messageRouter() {
	var err error
	defer this.transactionCleanup()
	for {
		select {
		case t := <-this.transactionChan:
			this.transactions = append(this.transactions, t)
			this.SetWriteDeadline(time.Now().Add(this.WriteTimeout))
			if err = t.cmd.Write(this.Conn); err != nil {
				log.Printf("[%s] error writing %s", this.RemoteAddr(), err.Error())
				this.Close()
				return
			}
		case buf := <-this.dataChan:
			frameType, data, err := UnpackResponse(buf)
			if err != nil {
				log.Printf("[%s] error (%s) unpacking response %d %s", this.RemoteAddr(), err.Error(), frameType, data)
				return
			}
			if frameType == FrameTypeResponse &&
				bytes.Equal(data, []byte("_heartbeat_")) {
				log.Printf("[%s] received heartbeat", this.RemoteAddr())
				if err := this.heartbeat(); err != nil {
					log.Printf("[%s] error sending heartbeat - %s", this.RemoteAddr(), err.Error())
					this.Close()
					return
				}
			} else {
				t := this.transactions[0]
				this.transactions = this.transactions[1:]
				t.frameType = frameType
				t.data = data
				t.err = err
				t.doneChan <- 1
			}
		case <-this.exitChan:
			return
		}
	}
}

//send heartbeat
func (this *Writer) heartbeat() error {
	this.SetWriteDeadline(time.Now().Add(this.WriteTimeout))
	if err := Nop().Write(this.Conn); err != nil {
		return err
	}
	return nil
Repository: nsqio/nsq
File: nsq/writer_test.go (Go)

Code before:
type ReaderHandler struct {
	t              *testing.T
	q              *Reader
	messagesGood   int
	messagesFailed int
}

func (h *ReaderHandler) LogFailedMessage(message *Message) {
	h.messagesFailed++
	h.q.Stop()
}

func (h *ReaderHandler) HandleMessage(message *Message) error {
	msg := string(message.Body)
	if msg == "bad_test_case" {
		return errors.New("fail this message")
	}
	if msg != "multipublish_test_case" && msg != "publish_test_case" {
		h.t.Error("message 'action' was not correct:", msg)
	}
	h.messagesGood++
	return nil
}
func TestConnection(t *testing.T) {
	addr := "127.0.0.1:4150"
	w := NewWriter()
	err := w.ConnectToNSQ(addr)
	if err != nil {
		t.Fatalf(err.Error())
	}
	w.Stop()
	cmd := Publish("write_test", []byte("failed test"))
	_, _, err = w.Write(cmd)
	if err.Error() != "not connected" {
		t.Fatalf("should not be able to write")
	}
	err = w.ConnectToNSQ(addr)
	if err.Error() != "writer stopped" {
		t.Fatalf("should not be able to connect to NSQ after Stop()")
	}
}

func TestWriteCommand(t *testing.T) {
	addr := "127.0.0.1:4150"
	topicName := "write_test" + strconv.Itoa(int(time.Now().Unix()))
	w := NewWriter()
	err := w.ConnectToNSQ(addr)
	if err != nil {
		t.Fatalf(err.Error())
	}
	msg_count := 10

Review comment (suggestion):
also, all the tests need to quiet log output:

``` go
log.SetOutput(ioutil.Discard)
defer log.SetOutput(io.Stdout)
```


Code after:
)

type ReaderHandler struct {
	t              *testing.T
	q              *Reader
	messagesGood   int
	messagesFailed int
}

func (h *ReaderHandler) LogFailedMessage(message *Message) {
	h.messagesFailed++
	h.q.Stop()
}

func (h *ReaderHandler) HandleMessage(message *Message) error {
	msg := string(message.Body)
	if msg == "bad_test_case" {
		return errors.New("fail this message")
	}
	if msg != "multipublish_test_case" && msg != "publish_test_case" {
		h.t.Error("message 'action' was not correct:", msg)
	}
	h.messagesGood++
	return nil
}
func TestConnection(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	addr := "127.0.0.1:4150"
	w := NewWriter(0)
	err := w.ConnectToNSQ(addr)
	if err != nil {
		t.Fatalf(err.Error())
	}
	w.Stop()
	_, _, err = w.Publish("write_test", []byte("failed test"))
	if err.Error() != "not connected" {
		t.Fatalf("should not be able to write")
	}
	err = w.ConnectToNSQ(addr)
	if err.Error() != "writer stopped" {
		t.Fatalf("should not be able to connect to NSQ after Stop()")
	}
}

func TestWriteCommand(t *testing.T) {
	log.SetOutput(ioutil.Discard)
	defer log.SetOutput(os.Stdout)

	addr := "127.0.0.1:4150"
Repository: nsqio/nsq
File: examples/nsq_to_file/nsq_to_file.go (Go)

Code before:
}

func (l *FileLogger) HandleMessage(m *nsq.Message, responseChannel chan *nsq.FinishedMessage) {
	l.logChan <- &Message{m, responseChannel}
}

func (f *FileLogger) router(r *nsq.Reader, termChan chan os.Signal, hupChan chan os.Signal) {
	pos := 0
	output := make([]*Message, *maxInFlight)
	sync := false
	ticker := time.NewTicker(time.Duration(30) * time.Second)
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					sync = true
					closeFile = true
				} else {
					if f.updateFile() {
						sync = true
					}
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {

Review comment (suggestion):
i dont think we need the `closing` var anymore at all, and I dont think we need to set sync = true here (it will be eventually set by the case above reading on the reader's `ExitChan`)


Code after:
}

func (l *FileLogger) HandleMessage(m *nsq.Message, responseChannel chan *nsq.FinishedMessage) {
	l.logChan <- &Message{m, responseChannel}
}

func (f *FileLogger) router(r *nsq.Reader, termChan chan os.Signal, hupChan chan os.Signal) {
	pos := 0
	output := make([]*Message, *maxInFlight)
	sync := false
	ticker := time.NewTicker(time.Duration(30) * time.Second)
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					closeFile = true
				} else {
					f.updateFile()
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {
				log.Fatalf("ERROR: writing newline to disk - %s", err.Error())
			}
			output[pos] = m
Repository: nsqio/nsq
File: examples/nsq_to_file/nsq_to_file.go (Go)

Code before:
}

func (l *FileLogger) HandleMessage(m *nsq.Message, responseChannel chan *nsq.FinishedMessage) {
	l.logChan <- &Message{m, responseChannel}
}

func (f *FileLogger) router(r *nsq.Reader, termChan chan os.Signal, hupChan chan os.Signal) {
	pos := 0
	output := make([]*Message, *maxInFlight)
	sync := false
	ticker := time.NewTicker(time.Duration(30) * time.Second)
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					sync = true
					closeFile = true
				} else {
					if f.updateFile() {
						sync = true
					}
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {

Review comment (suggestion):
The idea is to err on the side of the most durability possible. In the shutdown case, we flush what we have to disk (`sync`), and start the reader shutdown, at that point in time `closing` means we sync every additional record to disk one at a time, and respond to the server immediately (because there could still be in-flight messages we'll receive after this point in time).


Code after:
}

func (l *FileLogger) HandleMessage(m *nsq.Message, responseChannel chan *nsq.FinishedMessage) {
	l.logChan <- &Message{m, responseChannel}
}

func (f *FileLogger) router(r *nsq.Reader, termChan chan os.Signal, hupChan chan os.Signal) {
	pos := 0
	output := make([]*Message, *maxInFlight)
	sync := false
	ticker := time.NewTicker(time.Duration(30) * time.Second)
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					closeFile = true
				} else {
					f.updateFile()
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {
				log.Fatalf("ERROR: writing newline to disk - %s", err.Error())
			}
			output[pos] = m
Repository: nsqio/nsq
File: examples/nsq_to_file/nsq_to_file.go (Go)

Code before:
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					sync = true
					closeFile = true
				} else {
					if f.updateFile() {
						sync = true
					}
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {
				log.Fatalf("ERROR: writing newline to disk - %s", err.Error())
			}
			output[pos] = m
			pos++
			if pos == *maxInFlight {
				sync = true
			}
		}

		if closing || sync || r.IsStarved() {
			if pos > 0 {

Review comment (suggestion):
since we always set `sync = true` at the end of this case, it's functionally the same to write this as:

``` go
if f.needsFileRotate() {
    if *skipEmptyFiles {
        closeFile = true
    } else { 
        f.updateFile()
    }
}
sync = true
```


Code after:
	closing := false
	closeFile := false
	exit := false

	for {
		select {
		case <-r.ExitChan:
			sync = true
			closeFile = true
			exit = true
		case <-termChan:
			ticker.Stop()
			r.Stop()
			sync = true
			closing = true
		case <-hupChan:
			sync = true
			closeFile = true
		case <-ticker.C:
			if f.needsFileRotate() {
				if *skipEmptyFiles {
					closeFile = true
				} else {
					f.updateFile()
				}
			}
			sync = true
		case m := <-f.logChan:
			if f.updateFile() {
				sync = true
			}
			_, err := f.Write(m.Body)
			if err != nil {
				log.Fatalf("ERROR: writing message to disk - %s", err.Error())
			}
			_, err = f.Write([]byte("\n"))
			if err != nil {
				log.Fatalf("ERROR: writing newline to disk - %s", err.Error())
			}
			output[pos] = m
			pos++
			if pos == *maxInFlight {
				sync = true
			}
		}

		if closing || sync || r.IsStarved() {
			if pos > 0 {
				log.Printf("syncing %d records to disk", pos)
				err := f.Sync()
				if err != nil {
Repository: nsqio/nsq
File: nsqd/http.go (Go)

Code before:
		return
	}

	if !nsq.IsValidTopicName(topicName) {
		util.ApiResponse(w, 500, "INVALID_ARG_TOPIC", nil)
		return
	}

	if int64(len(reqParams.Body)) > s.context.nsqd.options.maxMessageSize {
		util.ApiResponse(w, 500, "MSG_TOO_BIG", nil)
		return
	}

	topic := s.context.nsqd.GetTopic(topicName)
	msg := nsq.NewMessage(<-s.context.nsqd.idChan, reqParams.Body)
	err = topic.PutMessage(msg)
	if err != nil {
		util.ApiResponse(w, 500, "NOK", nil)
		return
	}

	w.Header().Set("Content-Length", "2")
	io.WriteString(w, "OK")
}

func (s *httpServer) mputHandler(w http.ResponseWriter, req *http.Request) {
	var msgs []*nsq.Message
	var exit bool

	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		log.Printf("ERROR: failed to parse request params - %s", err.Error())
		util.ApiResponse(w, 500, "INVALID_REQUEST", nil)
	}

	topicNames, ok := reqParams["topic"]
	if !ok {
		util.ApiResponse(w, 500, "MISSING_ARG_TOPIC", nil)
		return
	}
	topicName := topicNames[0]

	if !nsq.IsValidTopicName(topicName) {
		util.ApiResponse(w, 500, "INVALID_ARG_TOPIC", nil)
		return
	}
	topic := s.context.nsqd.GetTopic(topicName)

	rdr := bufio.NewReader(req.Body)
	for !exit {
		block, err := rdr.ReadBytes('\n')

Review comment (suggestion):
Do you think we should keep an explicit check for `POST`? It would also be nice to keep some of the error messages the same between `/put` and `/mput`.


Code after:
		return
	}

	if !nsq.IsValidTopicName(topicName) {
		util.ApiResponse(w, 500, "INVALID_ARG_TOPIC", nil)
		return
	}

	if int64(len(reqParams.Body)) > s.context.nsqd.options.maxMessageSize {
		util.ApiResponse(w, 500, "MSG_TOO_BIG", nil)
		return
	}

	topic := s.context.nsqd.GetTopic(topicName)
	msg := nsq.NewMessage(<-s.context.nsqd.idChan, reqParams.Body)
	err = topic.PutMessage(msg)
	if err != nil {
		util.ApiResponse(w, 500, "NOK", nil)
		return
	}

	w.Header().Set("Content-Length", "2")
	io.WriteString(w, "OK")
}

func (s *httpServer) mputHandler(w http.ResponseWriter, req *http.Request) {
	var msgs []*nsq.Message
	var exit bool

	if req.Method != "POST" {
		util.ApiResponse(w, 500, "INVALID_REQUEST", nil)
		return
	}

	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		log.Printf("ERROR: failed to parse request params - %s", err.Error())
		util.ApiResponse(w, 500, "INVALID_REQUEST", nil)
	}

	topicNames, ok := reqParams["topic"]
	if !ok {
		util.ApiResponse(w, 500, "MISSING_ARG_TOPIC", nil)
		return
	}
	topicName := topicNames[0]

	if !nsq.IsValidTopicName(topicName) {
		util.ApiResponse(w, 500, "INVALID_ARG_TOPIC", nil)
		return
	}