Repository: nsqio/nsq
File: nsqd/protocol_v2.go (Go)

Code before:
		if err != nil {
			return nil, util.NewFatalClientErr(err, "E_BAD_MESSAGE",
				fmt.Sprintf("MPUB failed to read message(%d) body size", i))
		}

		if messageSize <= 0 {
			return nil, util.NewFatalClientErr(nil, "E_BAD_MESSAGE",
				fmt.Sprintf("MPUB invalid message(%d) body size %d", i, messageSize))
		}

		if int64(messageSize) > p.context.nsqd.options.maxMessageSize {
			return nil, util.NewFatalClientErr(nil, "E_BAD_MESSAGE",
				fmt.Sprintf("MPUB message too big %d > %d", messageSize, p.context.nsqd.options.maxMessageSize))
		}

		msgBody := make([]byte, messageSize)
		_, err = io.ReadFull(client.Reader, msgBody)
		if err != nil {
			return nil, util.NewFatalClientErr(err, "E_BAD_MESSAGE", "MPUB failed to read message body")
		}

		messages = append(messages, nsq.NewMessage(<-p.context.nsqd.idChan, msgBody))
	}

	topic := p.context.nsqd.GetTopic(topicName)

	// if we've made it this far we've validated all the input,
	// the only possible error is that the topic is exiting during
	// this next call (and no messages will be queued in that case)
	err = topic.PutMessages(messages)
	if err != nil {
		return nil, util.NewFatalClientErr(err, "E_MPUB_FAILED", "MPUB failed "+err.Error())
	}

	return okBytes, nil
}

func (p *ProtocolV2) TOUCH(client *ClientV2, params [][]byte) ([]byte, error) {
	state := atomic.LoadInt32(&client.State)
	if state != nsq.StateSubscribed && state != nsq.StateClosing {
		return nil, util.NewFatalClientErr(nil, "E_INVALID", "cannot TOUCH in current state")
	}

	if len(params) < 2 {
		return nil, util.NewFatalClientErr(nil, "E_INVALID", "TOUCH insufficient number of params")
	}

	id := *(*nsq.MessageID)(unsafe.Pointer(&params[1][0]))
	err := client.Channel.TouchMessage(client.ID, id)
	if err != nil {
		return nil, util.NewClientErr(err, "E_TOUCH_FAILED",

Review comment (suggestion):
this method was just pulled out of `MPUB` to be re-used in the HTTP code path, nothing functionally changed


Code after:
	}

	return okBytes, nil
}

func (p *ProtocolV2) TOUCH(client *ClientV2, params [][]byte) ([]byte, error) {
	state := atomic.LoadInt32(&client.State)
	if state != nsq.StateSubscribed && state != nsq.StateClosing {
		return nil, util.NewFatalClientErr(nil, "E_INVALID", "cannot TOUCH in current state")
	}

	if len(params) < 2 {
		return nil, util.NewFatalClientErr(nil, "E_INVALID", "TOUCH insufficient number of params")
	}

	id := *(*nsq.MessageID)(unsafe.Pointer(&params[1][0]))
	err := client.Channel.TouchMessage(client.ID, id)
	if err != nil {
		return nil, util.NewClientErr(err, "E_TOUCH_FAILED",
			fmt.Sprintf("TOUCH %s failed %s", id, err.Error()))
	}

	return nil, nil
}

func readMPUB(r io.Reader, tmp []byte, idChan chan nsq.MessageID, maxMessageSize int64) ([]*nsq.Message, error) {
	numMessages, err := readLen(r, tmp)
	if err != nil {
		return nil, util.NewFatalClientErr(err, "E_BAD_BODY", "MPUB failed to read message count")
	}

	if numMessages <= 0 {
		return nil, util.NewFatalClientErr(err, "E_BAD_BODY",
			fmt.Sprintf("MPUB invalid message count %d", numMessages))
	}

	messages := make([]*nsq.Message, 0, numMessages)
	for i := int32(0); i < numMessages; i++ {
		messageSize, err := readLen(r, tmp)
		if err != nil {
			return nil, util.NewFatalClientErr(err, "E_BAD_MESSAGE",
				fmt.Sprintf("MPUB failed to read message(%d) body size", i))
		}

		if messageSize <= 0 {
			return nil, util.NewFatalClientErr(nil, "E_BAD_MESSAGE",
				fmt.Sprintf("MPUB invalid message(%d) body size %d", i, messageSize))
		}

		if int64(messageSize) > maxMessageSize {
			return nil, util.NewFatalClientErr(nil, "E_BAD_MESSAGE",