func internalPubAsync(clientTimer *time.Timer, msgBody *bytes.Buffer, topic *nsqd.Topic) error {
	if topic.Exiting() {
		return nsqd.ErrExiting
	}
	info := &nsqd.PubInfo{
		Done:     make(chan struct{}),
		MsgBody:  msgBody,
		StartPub: time.Now(),
	}
	if clientTimer == nil {
		clientTimer = time.NewTimer(time.Second * 5)
	} else {
		clientTimer.Reset(time.Second * 5)
	}
	select {
	case topic.GetWaitChan() <- info:
	default:
		select {
		case topic.GetWaitChan() <- info:
		case <-topic.QuitChan():
			nsqd.NsqLogger().Infof("topic %v put messages failed at exiting", topic.GetFullName())
			return nsqd.ErrExiting
		case <-clientTimer.C:
			nsqd.NsqLogger().Infof("topic %v put messages timeout ", topic.GetFullName())
			return ErrPubToWaitTimeout
		}
	}
	<-info.Done
	return info.Err
}

func (s *httpServer) doStats(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failed to parse request params - %s", err)
		return nil, http_api.Err{400, "INVALID_REQUEST"}
	}
	formatString := reqParams.Get("format")
	topicName := reqParams.Get("topic")
	topicPart := reqParams.Get("partition")
	channelName := reqParams.Get("channel")
	leaderOnlyStr := reqParams.Get("leaderOnly")
	var leaderOnly bool
	leaderOnly, _ = strconv.ParseBool(leaderOnlyStr)

	jsonFormat := formatString == "json"

	stats := s.ctx.getStats(leaderOnly)
	health := s.ctx.getHealth()
	startTime := s.ctx.getStartTime()
	uptime := time.Since(startTime)

	// If we WERE given a topic-name, remove stats for all the other topics:
	if len(topicName) > 0 {
		// Find the desired-topic-index:
		for _, topicStats := range stats {
			if topicStats.TopicName == topicName {
				if len(topicPart) > 0 && topicStats.TopicPartition != topicPart {
					nsqd.NsqLogger().Logf("ignored stats topic partition mismatch - %v, %v", topicPart, topicStats.TopicPartition)
					continue
				}
				// If we WERE given a channel-name, remove stats for all the other channels:
				if len(channelName) > 0 {
					// Find the desired-channel:
					for _, channelStats := range topicStats.Channels {
						if channelStats.ChannelName == channelName {
							topicStats.Channels = []nsqd.ChannelStats{channelStats}
							// We've got the channel we were looking for:
							break
						}
					}
				}

				// We've got the topic we were looking for:
				stats = []nsqd.TopicStats{topicStats}
				break
			}
		}
	}

	if !jsonFormat {
		return s.printStats(stats, health, startTime, uptime), nil
	}

	return struct {
		Version   string            `json:"version"`
		Health    string            `json:"health"`
		StartTime int64             `json:"start_time"`
		Topics    []nsqd.TopicStats `json:"topics"`
	}{version.Binary, health, startTime.Unix(), stats}, nil
}

func handleRequestReponseForClient(client *nsqd.ClientV2, response []byte, err error) error {
	if err != nil {
		ctx := ""

		if childErr, ok := err.(protocol.ChildErr); ok {
			if parentErr := childErr.Parent(); parentErr != nil {
				ctx = " - " + parentErr.Error()
			}
		}

		nsqd.NsqLogger().LogDebugf("Error response for [%s] - %s - %s",
			client, err, ctx)

		sendErr := Send(client, frameTypeError, []byte(err.Error()))
		if sendErr != nil {
			nsqd.NsqLogger().LogErrorf("Send response error: [%s] - %s%s", client, sendErr, ctx)
			return err
		}

		// errors of type FatalClientErr should forceably close the connection
		if _, ok := err.(*protocol.FatalClientErr); ok {
			return err
		}
		return nil
	}

	if response != nil {
		sendErr := Send(client, frameTypeResponse, response)
		if sendErr != nil {
			err = fmt.Errorf("failed to send response - %s", sendErr)
		}
	}

	return err
}

func (s *httpServer) doEmptyChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	_, topic, channelName, err := s.getExistingTopicChannelFromQuery(req)
	if err != nil {
		return nil, err
	}

	channel, err := topic.GetExistingChannel(channelName)
	if err != nil {
		return nil, http_api.Err{404, "CHANNEL_NOT_FOUND"}
	}

	if s.ctx.checkForMasterWrite(topic.GetTopicName(), topic.GetTopicPart()) {
		var startFrom ConsumeOffset
		startFrom.OffsetType = offsetSpecialType
		startFrom.OffsetValue = -1
		queueOffset, cnt, err := s.ctx.SetChannelOffset(channel, &startFrom, true)
		if err != nil {
			return nil, http_api.Err{500, err.Error()}
		}
		nsqd.NsqLogger().Logf("empty the channel to end offset: %v:%v, by client:%v",
			queueOffset, cnt, req.RemoteAddr)
	} else {
		nsqd.NsqLogger().LogDebugf("should request to master: %v, from %v",
			topic.GetFullName(), req.RemoteAddr)
		return nil, http_api.Err{400, FailedOnNotLeader}
	}
	return nil, nil
}

func (s *httpServer) enableMessageTrace(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failed to parse request params - %s", err)
		return nil, http_api.Err{400, "INVALID_REQUEST"}
	}
	topicName := reqParams.Get("topic")
	channelName := reqParams.Get("channel")

	parts := s.ctx.getPartitions(topicName)
	for _, t := range parts {
		if channelName != "" {
			ch, err := t.GetExistingChannel(channelName)
			if err != nil {
				continue
			}
			ch.SetTrace(true)
			nsqd.NsqLogger().Logf("channel %v trace enabled", ch.GetName())
		} else {
			t.SetTrace(true)
			nsqd.NsqLogger().Logf("topic %v trace enabled", t.GetFullName())
		}
	}
	return nil, nil
}

func (n *NsqdServer) discoverLookupdNodes(discoveryAddrs []string) ([]string, bool) {
	changed := false
	// discovery the new lookup
	if n.ctx.nsqdCoord != nil {
		newDiscoveried, err := n.ctx.nsqdCoord.GetAllLookupdNodes()
		if err != nil {
			nsqd.NsqLogger().Logf("discovery lookup failed: %v", err)
		} else {
			if len(newDiscoveried) != len(discoveryAddrs) {
				changed = true
			} else {
				for _, l := range newDiscoveried {
					if !in(net.JoinHostPort(l.NodeIP, l.TcpPort), discoveryAddrs) {
						changed = true
						break
					}
				}
			}
			if changed {
				discoveryAddrs = discoveryAddrs[:0]
				for _, l := range newDiscoveried {
					discoveryAddrs = append(discoveryAddrs, net.JoinHostPort(l.NodeIP, l.TcpPort))
				}
				nsqd.NsqLogger().LogDebugf("discovery lookup nodes: %v", discoveryAddrs)
			}
		}
	}
	return discoveryAddrs, changed
}

func (s *httpServer) getExistingTopicFromQuery(req *http.Request) (url.Values, *nsqd.Topic, error) {
	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failed to parse request params - %s", err)
		return nil, nil, http_api.Err{400, "INVALID_REQUEST"}
	}

	topicName, topicPart, err := http_api.GetTopicPartitionArgs(reqParams)
	if err != nil {
		return nil, nil, http_api.Err{400, err.Error()}
	}

	if topicPart == -1 {
		topicPart = s.ctx.getDefaultPartition(topicName)
	}

	topic, err := s.ctx.getExistingTopic(topicName, topicPart)
	if err != nil {
		nsqd.NsqLogger().Logf("topic not found - %s-%v, %v", topicName, topicPart, err)
		return nil, nil, http_api.Err{404, E_TOPIC_NOT_EXIST}
	}

	if topicPart != topic.GetTopicPart() {
		return nil, nil, http_api.Err{http.StatusNotFound, "Topic partition not exist"}
	}

	return reqParams, topic, nil
}

func newHTTPServer(ctx *context, tlsEnabled bool, tlsRequired bool) *httpServer {
	log := http_api.Log(nsqd.NsqLogger())

	router := httprouter.New()
	router.HandleMethodNotAllowed = true
	router.PanicHandler = http_api.LogPanicHandler(nsqd.NsqLogger())
	router.NotFound = http_api.LogNotFoundHandler(nsqd.NsqLogger())
	router.MethodNotAllowed = http_api.LogMethodNotAllowedHandler(nsqd.NsqLogger())
	s := &httpServer{
		ctx:         ctx,
		tlsEnabled:  tlsEnabled,
		tlsRequired: tlsRequired,
		router:      router,
	}

	router.Handle("GET", "/ping", http_api.Decorate(s.pingHandler, log, http_api.PlainText))
	router.Handle("POST", "/loglevel/set", http_api.Decorate(s.doSetLogLevel, log, http_api.V1))
	router.Handle("GET", "/info", http_api.Decorate(s.doInfo, log, http_api.NegotiateVersion))

	// v1 negotiate
	router.Handle("POST", "/pub", http_api.Decorate(s.doPUB, http_api.NegotiateVersion))
	router.Handle("POST", "/pubtrace", http_api.Decorate(s.doPUBTrace, http_api.V1))
	router.Handle("POST", "/mpub", http_api.Decorate(s.doMPUB, http_api.NegotiateVersion))
	router.Handle("GET", "/stats", http_api.Decorate(s.doStats, log, http_api.NegotiateVersion))
	router.Handle("GET", "/coordinator/stats", http_api.Decorate(s.doCoordStats, log, http_api.V1))
	router.Handle("GET", "/message/stats", http_api.Decorate(s.doMessageStats, log, http_api.V1))
	router.Handle("GET", "/message/historystats", http_api.Decorate(s.doMessageHistoryStats, log, http_api.V1))
	router.Handle("POST", "/message/trace/enable", http_api.Decorate(s.enableMessageTrace, log, http_api.V1))
	router.Handle("POST", "/message/trace/disable", http_api.Decorate(s.disableMessageTrace, log, http_api.V1))
	router.Handle("POST", "/channel/pause", http_api.Decorate(s.doPauseChannel, log, http_api.V1))
	router.Handle("POST", "/channel/unpause", http_api.Decorate(s.doPauseChannel, log, http_api.V1))
	router.Handle("POST", "/channel/create", http_api.Decorate(s.doCreateChannel, log, http_api.V1))
	router.Handle("POST", "/channel/delete", http_api.Decorate(s.doDeleteChannel, log, http_api.V1))
	router.Handle("POST", "/channel/empty", http_api.Decorate(s.doEmptyChannel, log, http_api.V1))
	router.Handle("POST", "/channel/setoffset", http_api.Decorate(s.doSetChannelOffset, log, http_api.V1))
	router.Handle("POST", "/channel/setorder", http_api.Decorate(s.doSetChannelOrder, log, http_api.V1))
	router.Handle("GET", "/config/:opt", http_api.Decorate(s.doConfig, log, http_api.V1))
	router.Handle("PUT", "/config/:opt", http_api.Decorate(s.doConfig, log, http_api.V1))

	//router.Handle("POST", "/topic/delete", http_api.Decorate(s.doDeleteTopic, http_api.DeprecatedAPI, log, http_api.V1))

	// debug
	router.HandlerFunc("GET", "/debug/pprof/", pprof.Index)
	router.HandlerFunc("GET", "/debug/pprof/cmdline", pprof.Cmdline)
	router.HandlerFunc("GET", "/debug/pprof/symbol", pprof.Symbol)
	router.HandlerFunc("POST", "/debug/pprof/symbol", pprof.Symbol)
	router.HandlerFunc("GET", "/debug/pprof/profile", pprof.Profile)
	router.HandlerFunc("GET", "/debug/pprof/trace", pprof.Trace)
	router.Handler("GET", "/debug/pprof/heap", pprof.Handler("heap"))
	router.Handler("GET", "/debug/pprof/goroutine", pprof.Handler("goroutine"))
	router.Handler("GET", "/debug/pprof/block", pprof.Handler("block"))
	router.Handle("PUT", "/debug/setblockrate", http_api.Decorate(setBlockRateHandler, log, http_api.V1))
	router.Handler("GET", "/debug/pprof/threadcreate", pprof.Handler("threadcreate"))

	return s
}

func (p *protocolV2) internalMPUBAndTrace(client *nsqd.ClientV2, params [][]byte, traceEnable bool) ([]byte, error) {
	startPub := time.Now().UnixNano()
	_, topic, preErr := p.preparePub(client, params, p.ctx.getOpts().MaxBodySize)
	if preErr != nil {
		return nil, preErr
	}

	messages, buffers, preErr := readMPUB(client.Reader, client.LenSlice, topic,
		p.ctx.getOpts().MaxMsgSize, traceEnable)

	defer func() {
		for _, b := range buffers {
			topic.BufferPoolPut(b)
		}
	}()
	if preErr != nil {
		return nil, preErr
	}

	topicName := topic.GetTopicName()
	partition := topic.GetTopicPart()
	if p.ctx.checkForMasterWrite(topicName, partition) {
		id, offset, rawSize, err := p.ctx.PutMessages(topic, messages)
		//p.ctx.setHealth(err)
		if err != nil {
			topic.GetDetailStats().UpdatePubClientStats(client.RemoteAddr().String(), client.UserAgent, "tcp", int64(len(messages)), true)
			nsqd.NsqLogger().LogErrorf("topic %v put message failed: %v", topic.GetFullName(), err)

			if clusterErr, ok := err.(*consistence.CommonCoordErr); ok {
				if !clusterErr.IsLocalErr() {
					return nil, protocol.NewClientErr(err, FailedOnNotWritable, "")
				}
			}
			return nil, protocol.NewFatalClientErr(err, "E_MPUB_FAILED", err.Error())
		}
		topic.GetDetailStats().UpdatePubClientStats(client.RemoteAddr().String(), client.UserAgent, "tcp", int64(len(messages)), false)
		cost := time.Now().UnixNano() - startPub
		topic.GetDetailStats().UpdateTopicMsgStats(0, cost/1000/int64(len(messages)))
		if !traceEnable {
			return okBytes, nil
		}
		return getTracedReponse(buffers[0], id, 0, offset, rawSize)
	} else {
		topic.GetDetailStats().UpdatePubClientStats(client.RemoteAddr().String(), client.UserAgent, "tcp", int64(len(messages)), true)
		//forward to master of topic
		nsqd.NsqLogger().LogDebugf("should put to master: %v, from %v",
			topic.GetFullName(), client.RemoteAddr)
		topic.DisableForSlave()
		return nil, protocol.NewClientErr(preErr, FailedOnNotLeader, "")
	}
}

func (s *httpServer) doSetChannelOffset(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	_, topic, channelName, err := s.getExistingTopicChannelFromQuery(req)
	if err != nil {
		return nil, err
	}

	channel, err := topic.GetExistingChannel(channelName)
	if err != nil {
		return nil, http_api.Err{404, "CHANNEL_NOT_FOUND"}
	}
	readMax := req.ContentLength + 1
	body := make([]byte, req.ContentLength)
	n, err := io.ReadFull(io.LimitReader(req.Body, readMax), body)
	if err != nil {
		nsqd.NsqLogger().Logf("read request body error: %v", err)
		body = body[:n]
		if err == io.EOF || err == io.ErrUnexpectedEOF {
			// we ignore EOF, maybe the ContentLength is not match?
			nsqd.NsqLogger().LogWarningf("read request body eof: %v, ContentLength: %v,return length %v.",
				err, req.ContentLength, n)
		} else {
			return nil, http_api.Err{500, "INTERNAL_ERROR"}
		}
	}
	if len(body) == 0 {
		return nil, http_api.Err{406, "MSG_EMPTY"}
	}
	startFrom := &ConsumeOffset{}
	err = startFrom.FromBytes(body)
	if err != nil {
		nsqd.NsqLogger().Logf("offset %v error: %v", string(body), err)
		return nil, http_api.Err{400, err.Error()}
	}

	if s.ctx.checkForMasterWrite(topic.GetTopicName(), topic.GetTopicPart()) {
		queueOffset, cnt, err := s.ctx.SetChannelOffset(channel, startFrom, true)
		if err != nil {
			return nil, http_api.Err{500, err.Error()}
		}
		nsqd.NsqLogger().Logf("set the channel offset: %v (actual set : %v:%v), by client:%v",
			startFrom, queueOffset, cnt, req.RemoteAddr)
	} else {
		nsqd.NsqLogger().LogDebugf("should request to master: %v, from %v",
			topic.GetFullName(), req.RemoteAddr)
		return nil, http_api.Err{400, FailedOnNotLeader}
	}
	return nil, nil
}

func (p *protocolV2) TOUCH(client *nsqd.ClientV2, params [][]byte) ([]byte, error) {
	state := atomic.LoadInt32(&client.State)
	if state != stateSubscribed && state != stateClosing {
		nsqd.NsqLogger().LogWarningf("[%s] command in wrong state: %v", client, state)
		return nil, protocol.NewFatalClientErr(nil, E_INVALID, "cannot TOUCH in current state")
	}

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

	id, err := getFullMessageID(params[1])
	if err != nil {
		return nil, protocol.NewFatalClientErr(nil, E_INVALID, err.Error())
	}

	client.LockRead()
	msgTimeout := client.MsgTimeout
	client.UnlockRead()

	if client.Channel == nil {
		return nil, protocol.NewFatalClientErr(nil, E_INVALID, "No channel")
	}
	err = client.Channel.TouchMessage(client.ID, nsqd.GetMessageIDFromFullMsgID(*id), msgTimeout)
	if err != nil {
		return nil, protocol.NewClientErr(err, "E_TOUCH_FAILED",
			fmt.Sprintf("TOUCH %v failed %s", *id, err.Error()))
	}

	return nil, nil
}

func (s *httpServer) doPauseChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	_, topic, channelName, err := s.getExistingTopicChannelFromQuery(req)
	if err != nil {
		return nil, err
	}

	channel, err := topic.GetExistingChannel(channelName)
	if err != nil {
		return nil, http_api.Err{404, "CHANNEL_NOT_FOUND"}
	}

	if strings.Contains(req.URL.Path, "unpause") {
		err = channel.UnPause()
	} else {
		err = channel.Pause()
	}
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failure in %s - %s", req.URL.Path, err)
		return nil, http_api.Err{500, "INTERNAL_ERROR"}
	}

	// pro-actively persist metadata so in case of process failure
	s.ctx.persistMetadata()
	return nil, nil
}

func connectCallback(ctx *context, hostname string, syncTopicChan chan *clusterinfo.LookupPeer, exitChan chan int) func(*clusterinfo.LookupPeer) {
	return func(lp *clusterinfo.LookupPeer) {
		ci := make(map[string]interface{})
		ci["version"] = version.Binary
		ci["tcp_port"] = ctx.realTCPAddr().Port
		if ctx.reverseProxyPort == "" {
			ci["http_port"] = ctx.realHTTPAddr().Port
		} else {
			port, _ := strconv.Atoi(ctx.reverseProxyPort)
			ci["http_port"] = port
		}
		ci["hostname"] = hostname
		ci["broadcast_address"] = ctx.getOpts().BroadcastAddress
		ci["distributed_id"] = ctx.GetDistributedID()

		cmd, err := nsq.Identify(ci)
		if err != nil {
			lp.Close()
			return
		}
		resp, err := lp.Command(cmd)
		if err != nil {
			nsqd.NsqLogger().Logf("LOOKUPD(%s): ERROR %s - %s", lp, cmd, err)
		} else if bytes.Equal(resp, []byte("E_INVALID")) {
			nsqd.NsqLogger().Logf("LOOKUPD(%s): lookupd returned %s", lp, resp)
		} else {
			err = json.Unmarshal(resp, &lp.Info)
			if err != nil {
				nsqd.NsqLogger().Logf("LOOKUPD(%s): ERROR parsing response - %s", lp, resp)
			} else {
				nsqd.NsqLogger().Logf("LOOKUPD(%s): peer info %+v", lp, lp.Info)
			}
		}

		go func() {
			select {
			case syncTopicChan <- lp:
			case <-exitChan:
				return
			}
		}()
	}
}

func NewNsqdServer(opts *nsqd.Options) (*nsqd.NSQD, *NsqdServer) {
	ip := opts.DecideBroadcast()

	nsqdInstance := nsqd.New(opts)

	s := &NsqdServer{}
	ctx := &context{}
	ctx.nsqd = nsqdInstance
	_, port, _ := net.SplitHostPort(opts.TCPAddress)
	rpcport := opts.RPCPort
	if rpcport != "" {
		ip = opts.BroadcastAddress
		consistence.SetCoordLogger(opts.Logger, opts.LogLevel)
		coord := consistence.NewNsqdCoordinator(opts.ClusterID, ip, port, rpcport, strconv.FormatInt(opts.ID, 10), opts.DataPath, nsqdInstance)
		l := consistence.NewNsqdEtcdMgr(opts.ClusterLeadershipAddresses)
		coord.SetLeadershipMgr(l)
		ctx.nsqdCoord = coord
	} else {
		nsqd.NsqLogger().LogWarningf("Start without nsqd coordinator enabled")
		ctx.nsqdCoord = nil
	}

	s.ctx = ctx

	s.exitChan = make(chan int)

	tlsConfig, err := buildTLSConfig(opts)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("FATAL: failed to build TLS config - %s", err)
		os.Exit(1)
	}
	if tlsConfig == nil && opts.TLSRequired != TLSNotRequired {
		nsqd.NsqLogger().LogErrorf("FATAL: cannot require TLS client connections without TLS key and cert")
		os.Exit(1)
	}
	s.ctx.tlsConfig = tlsConfig
	s.ctx.nsqd.SetPubLoop(s.ctx.internalPubLoop)

	nsqd.NsqLogger().Logf(version.String("nsqd"))
	nsqd.NsqLogger().Logf("ID: %d", opts.ID)

	return nsqdInstance, s
}

func (p *protocolV2) CLS(client *nsqd.ClientV2, params [][]byte) ([]byte, error) {
	state := atomic.LoadInt32(&client.State)
	if state != stateSubscribed {
		nsqd.NsqLogger().LogWarningf("[%s] command in wrong state: %v", client, state)
		return nil, protocol.NewFatalClientErr(nil, E_INVALID, "cannot CLS in current state")
	}

	client.StartClose()

	return []byte("CLOSE_WAIT"), nil
}

func (s *httpServer) doDeleteChannel(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	_, topic, channelName, err := s.getExistingTopicChannelFromQuery(req)
	if err != nil {
		return nil, err
	}

	if s.ctx.checkForMasterWrite(topic.GetTopicName(), topic.GetTopicPart()) {
		clusterErr := s.ctx.DeleteExistingChannel(topic, channelName)
		if clusterErr != nil {
			return nil, http_api.Err{500, clusterErr.Error()}
		}
		nsqd.NsqLogger().Logf("deleted the channel : %v, by client:%v",
			channelName, req.RemoteAddr)
	} else {
		nsqd.NsqLogger().LogDebugf("should request to master: %v, from %v",
			topic.GetFullName(), req.RemoteAddr)
		return nil, http_api.Err{400, FailedOnNotLeader}
	}

	return nil, nil
}

func (p *protocolV2) preparePub(client *nsqd.ClientV2, params [][]byte, maxBody int64) (int32, *nsqd.Topic, error) {
	var err error

	if len(params) < 2 {
		return 0, nil, protocol.NewFatalClientErr(nil, E_INVALID, "insufficient number of parameters")
	}

	topicName := string(params[1])
	if !protocol.IsValidTopicName(topicName) {
		return 0, nil, protocol.NewFatalClientErr(nil, "E_BAD_TOPIC",
			fmt.Sprintf("topic name %q is not valid", topicName))
	}
	partition := -1
	if len(params) == 3 {
		partition, err = strconv.Atoi(string(params[2]))
		if err != nil {
			return 0, nil, protocol.NewFatalClientErr(nil, "E_BAD_PARTITION",
				fmt.Sprintf("topic partition is not valid: %v", err))
		}
	}

	if partition == -1 {
		partition = p.ctx.getDefaultPartition(topicName)
	}

	bodyLen, err := readLen(client.Reader, client.LenSlice)
	if err != nil {
		return 0, nil, protocol.NewFatalClientErr(err, "E_BAD_BODY", "failed to read body size")
	}

	if bodyLen <= 0 {
		return bodyLen, nil, protocol.NewFatalClientErr(nil, "E_BAD_BODY",
			fmt.Sprintf("invalid body size %d", bodyLen))
	}

	if int64(bodyLen) > maxBody {
		return bodyLen, nil, protocol.NewFatalClientErr(nil, "E_BAD_BODY",
			fmt.Sprintf("body too big %d > %d", bodyLen, maxBody))
	}

	topic, err := p.ctx.getExistingTopic(topicName, partition)
	if err != nil {
		nsqd.NsqLogger().Logf("not existing topic: %v-%v, err:%v", topicName, partition, err.Error())
		return bodyLen, nil, protocol.NewFatalClientErr(nil, E_TOPIC_NOT_EXIST, "")
	}

	if err := p.CheckAuth(client, "PUB", topicName, ""); err != nil {
		return bodyLen, nil, err
	}
	// mpub
	return bodyLen, topic, nil
}

func (s *httpServer) doCoordStats(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	if s.ctx.nsqdCoord != nil {
		reqParams, err := url.ParseQuery(req.URL.RawQuery)
		if err != nil {
			nsqd.NsqLogger().LogErrorf("failed to parse request params - %s", err)
			return nil, http_api.Err{400, "INVALID_REQUEST"}
		}
		topicName := reqParams.Get("topic")
		topicPartStr := reqParams.Get("partition")
		topicPart := -1
		if topicPartStr != "" {
			topicPart, err = strconv.Atoi(topicPartStr)
			if err != nil {
				nsqd.NsqLogger().LogErrorf("invalid partition: %v - %s", topicPartStr, err)
				return nil, http_api.Err{400, "INVALID_REQUEST"}
			}
		}

		return s.ctx.nsqdCoord.Stats(topicName, topicPart), nil
	}
	return nil, http_api.Err{500, "Coordinator is disabled."}
}

func (s *httpServer) doMessageStats(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
	reqParams, err := url.ParseQuery(req.URL.RawQuery)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failed to parse request params - %s", err)
		return nil, http_api.Err{400, "INVALID_REQUEST"}
	}
	topicName := reqParams.Get("topic")
	topicPartStr := reqParams.Get("partition")
	topicPart, err := strconv.Atoi(topicPartStr)
	if err != nil {
		nsqd.NsqLogger().LogErrorf("failed to get partition - %s", err)
		return nil, http_api.Err{400, "INVALID_REQUEST"}
	}
	channelName := reqParams.Get("channel")

	t, err := s.ctx.getExistingTopic(topicName, topicPart)
	if err != nil {
		return nil, http_api.Err{404, E_TOPIC_NOT_EXIST}
	}
	statStr := t.GetTopicChannelDebugStat(channelName)

	return statStr, nil
}

func (p *protocolV2) RDY(client *nsqd.ClientV2, params [][]byte) ([]byte, error) {
	state := atomic.LoadInt32(&client.State)

	if state == stateClosing {
		// just ignore ready changes on a closing channel
		nsqd.NsqLogger().Logf(
			"PROTOCOL(V2): [%s] ignoring RDY after CLS in state ClientStateV2Closing",
			client)
		return nil, nil
	}

	if state != stateSubscribed {
		nsqd.NsqLogger().LogWarningf("[%s] command in wrong state: %v", client, state)
		return nil, protocol.NewFatalClientErr(nil, E_INVALID, "cannot RDY in current state")
	}

	count := int64(1)
	if len(params) > 1 {
		b10, err := protocol.ByteToBase10(params[1])
		if err != nil {
			return nil, protocol.NewFatalClientErr(err, E_INVALID,
				fmt.Sprintf("RDY could not parse count %s", params[1]))
		}
		count = int64(b10)
	}

	if count < 0 || count > p.ctx.getOpts().MaxRdyCount {
		// this needs to be a fatal error otherwise clients would have
		// inconsistent state
		return nil, protocol.NewFatalClientErr(nil, E_INVALID,
			fmt.Sprintf("RDY count %d out of range 0-%d", count, p.ctx.getOpts().MaxRdyCount))
	}

	client.SetReadyCount(count)

	return nil, nil
}