gofmt -s

apps/nsq_data_tool/tool.go

@@ -102,6 +102,10 @@ func main() {
 		}
 		rets := make([]nsqd.Message, *viewCnt)
 		cnt, err := delayQ.PeekRecentChannelTimeout(time.Now().UnixNano(), rets, *viewCh)
+		if err != nil {
+			nsqd.NsqLogger().Infof("peek recent timeout failed: %v", err.Error())
+			return
+		}
 		for _, m := range rets[:cnt] {
 			nsqd.NsqLogger().Infof("peeked channel msg : %v", m)
 		}

apps/nsq_to_http/nsq_to_http.go

@@ -19,11 +19,11 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/bitly/go-hostpool"
+	"github.com/bitly/timer_metrics"
 	"github.com/youzan/go-nsq"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/version"
-	"github.com/bitly/go-hostpool"
-	"github.com/bitly/timer_metrics"
 )
 
 const (

apps/nsq_to_nsq/nsq_to_nsq.go

@@ -16,13 +16,13 @@ import (
 	"syscall"
 	"time"
 
+	"github.com/bitly/go-hostpool"
+	"github.com/bitly/go-simplejson"
+	"github.com/bitly/timer_metrics"
 	"github.com/youzan/go-nsq"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/protocol"
 	"github.com/youzan/nsq/internal/version"
-	"github.com/bitly/go-hostpool"
-	"github.com/bitly/go-simplejson"
-	"github.com/bitly/timer_metrics"
 )
 
 const (

apps/nsqadmin/main.go

@@ -11,10 +11,10 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/absolute8511/glog"
+	"github.com/mreiferson/go-options"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/version"
 	"github.com/youzan/nsq/nsqadmin"
-	"github.com/mreiferson/go-options"
 )
 
 var (

apps/nsqd/nsqd.go

@@ -15,12 +15,12 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/absolute8511/glog"
+	"github.com/judwhite/go-svc/svc"
+	"github.com/mreiferson/go-options"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/version"
 	"github.com/youzan/nsq/nsqd"
 	"github.com/youzan/nsq/nsqdserver"
-	"github.com/judwhite/go-svc/svc"
-	"github.com/mreiferson/go-options"
 )
 
 type tlsRequiredOption int

apps/nsqd/nsqd_test.go

@@ -6,8 +6,8 @@ import (
 	"testing"
 
 	"github.com/BurntSushi/toml"
-	"github.com/youzan/nsq/nsqd"
 	"github.com/mreiferson/go-options"
+	"github.com/youzan/nsq/nsqd"
 )
 
 func TestConfigFlagParsing(t *testing.T) {

apps/nsqlookupd/nsqlookupd.go

@@ -11,11 +11,11 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/absolute8511/glog"
+	"github.com/judwhite/go-svc/svc"
+	"github.com/mreiferson/go-options"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/version"
 	"github.com/youzan/nsq/nsqlookupd"
-	"github.com/judwhite/go-svc/svc"
-	"github.com/mreiferson/go-options"
 )
 
 var (

bench/multi_bench/multi_bench.go

@@ -17,12 +17,12 @@ import (
 	"time"
 
 	"github.com/absolute8511/glog"
+	"github.com/spaolacci/murmur3"
 	"github.com/youzan/go-nsq"
 	"github.com/youzan/nsq/internal/app"
 	"github.com/youzan/nsq/internal/clusterinfo"
 	"github.com/youzan/nsq/internal/http_api"
 	"github.com/youzan/nsq/internal/levellogger"
-	"github.com/spaolacci/murmur3"
 )
 
 var (
@@ -454,7 +454,7 @@ func startCheckData2() {
 			continue
 		}
 		log.Printf("topic: %v sub waiting list: %v\n", topicName, len(waitingList))
-		for traceID, _ := range waitingList {
+		for traceID := range waitingList {
 			log.Printf("%v, ", traceID)
 		}
 		log.Printf("\n")
@@ -1216,7 +1216,7 @@ func (c *consumeTraceIDHandler) HandleMessage(message *nsq.Message) error {
 	}
 	c.subTraceWaiting[traceID] = message
 	newMaxTraceID := atomic.LoadInt64(c.subIDCounter)
-	for fid, _ := range c.failedList {
+	for fid := range c.failedList {
 		// we treat as pub failed id already
 		if fid > uint64(newMaxTraceID) {
 			c.subTraceWaiting[fid] = nil
@@ -1235,7 +1235,7 @@ func (c *consumeTraceIDHandler) HandleMessage(message *nsq.Message) error {
 			break
 		}
 	}
-	for traceID, _ := range c.subTraceWaiting {
+	for traceID := range c.subTraceWaiting {
 		if traceID <= uint64(newMaxTraceID) {
 			delete(c.subTraceWaiting, traceID)
 		}

consistence/coordinator_rpc.go

@@ -604,9 +604,9 @@ type RpcChannelOffsetArg struct {
 
 type RpcPutMessages struct {
 	RpcTopicData
-	LogData         CommitLogData
-	TopicMessages   []*nsqd.Message
-	// raw message data will include the size header 
+	LogData       CommitLogData
+	TopicMessages []*nsqd.Message
+	// raw message data will include the size header
 	TopicRawMessage []byte
 }
 

consistence/data_placement_mgr.go

@@ -336,7 +336,7 @@ func (s topicLFListT) Less(i, j int) bool {
 
 func (self *NodeTopicStats) GetSortedTopicWriteLevel(leaderOnly bool) topicLFListT {
 	topicLFList := make(topicLFListT, 0)
-	for topicName, _ := range self.TopicHourlyPubDataList {
+	for topicName := range self.TopicHourlyPubDataList {
 		d, ok := self.TopicLeaderDataSize[topicName]
 		if leaderOnly {
 			if !ok || d <= 0 {
@@ -364,7 +364,7 @@ func (self *NodeTopicStats) GetMostBusyAndIdleTopicWriteLevel(leaderOnly bool) (
 	busyTopic := ""
 	idle := float64(math.MaxInt32)
 	idleTopic := ""
-	for topicName, _ := range self.TopicHourlyPubDataList {
+	for topicName := range self.TopicHourlyPubDataList {
 		d, ok := self.TopicLeaderDataSize[topicName]
 		if leaderOnly {
 			if !ok || d <= 0 {
@@ -1720,7 +1720,7 @@ func (self *DataPlacement) getRebalancedOrderedTopicPartitions(
 	// p9       l       x        f     x-f
 	// p10     x-f      x       f-l     f
 	nodeNameList := make(SortableStrings, 0, len(currentNodes))
-	for nid, _ := range currentNodes {
+	for nid := range currentNodes {
 		nodeNameList = append(nodeNameList, nid)
 	}
 	return self.getRebalancedOrderedTopicPartitionsFromNameList(partitionNum, replica, nodeNameList)

consistence/nsq_lookupd_etcd.go

@@ -16,8 +16,8 @@ import (
 	"sync/atomic"
 	"time"
 
-	"github.com/coreos/etcd/client"
 	etcdlock "github.com/absolute8511/xlock2"
+	"github.com/coreos/etcd/client"
 	"golang.org/x/net/context"
 )
 
@@ -194,7 +194,7 @@ func (self *NsqLookupdEtcdMgr) AcquireAndWatchLeader(leader chan *NsqLookupdNode
 	master.Start()
 }
 
-func  (self *NsqLookupdEtcdMgr) GetTopicsMetaInfoMap(topics []string) (map[string]*TopicMetaInfo, error) {
+func (self *NsqLookupdEtcdMgr) GetTopicsMetaInfoMap(topics []string) (map[string]*TopicMetaInfo, error) {
 	topicMetaInfoCache := make(map[string]*TopicMetaInfo)
 	if atomic.LoadInt32(&self.ifTopicChanged) == 1 {
 		//fetch from etcd
@@ -214,12 +214,12 @@ func  (self *NsqLookupdEtcdMgr) GetTopicsMetaInfoMap(topics []string) (map[strin
 				coordLog.Infof("meta info for %v not exist", topic)
 				topicMetaInfoCache[topic] = &TopicMetaInfo{
 					OrderedMulti: false,
-					Ext: false,
+					Ext:          false,
 				}
 			} else {
 				topicMetaInfoCache[topic] = &TopicMetaInfo{
 					OrderedMulti: topicMeta.OrderedMulti,
-					Ext: topicMeta.Ext,
+					Ext:          topicMeta.Ext,
 				}
 			}
 		}

consistence/nsqd_coordinator.go

@@ -494,7 +494,7 @@ func (self *NsqdCoordinator) getLookupRemoteProxy() (INsqlookupRemoteProxy, *Coo
 		self.lookupMutex.Lock()
 		self.lookupRemoteClients[addr] = c
 		if len(self.lookupRemoteClients) > 3 {
-			for k, _ := range self.lookupRemoteClients {
+			for k := range self.lookupRemoteClients {
 				if k == addr {
 					continue
 				}
@@ -960,7 +960,7 @@ func (self *NsqdCoordinator) checkForUnsyncedTopics() {
 		}
 		self.coordMutex.Unlock()
 		for topic, info := range tmpChecks {
-			for pid, _ := range info {
+			for pid := range info {
 				topicMeta, err := self.leadership.GetTopicInfo(topic, pid)
 				if err != nil {
 					continue
@@ -1822,7 +1822,7 @@ func (self *NsqdCoordinator) catchupFromLeader(topicInfo TopicPartitionMetaInfo,
 					delete(oldChList, chName)
 				}
 				coordLog.Infof("topic %v local channel not on leader: %v", topicInfo.GetTopicDesp(), oldChList)
-				for chName, _ := range oldChList {
+				for chName := range oldChList {
 					localTopic.CloseExistingChannel(chName, false)
 				}
 				localTopic.SaveChannelMeta()

consistence/nsqd_coordinator_cluster_write.go

@@ -458,7 +458,7 @@ retrysync:
 				// isr may down or some error.
 				// We also need do some work to decide if we
 				// should give up my leadership.
-				for nid, _ := range failedNodes {
+				for nid := range failedNodes {
 					tmpErr := self.requestLeaveFromISRByLeader(topicName, topicPartition, nid)
 					if tmpErr != nil {
 						coordLog.Warningf("failed to request remove the failed isr node: %v, %v", nid, tmpErr)

consistence/nsqd_node_etcd.go

@@ -14,8 +14,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/coreos/etcd/client"
 	etcdlock "github.com/absolute8511/xlock2"
+	"github.com/coreos/etcd/client"
 	"golang.org/x/net/context"
 )
 

consistence/nsqlookup_coordinator.go

@@ -489,7 +489,7 @@ func (self *NsqLookupCoordinator) handleRemovingNodes(monitorChan chan struct{})
 			}
 			By(leaderSort).Sort(nodeTopicStats)
 
-			for nid, _ := range removingNodes {
+			for nid := range removingNodes {
 				anyPending := false
 				coordLog.Infof("handle the removing node %v ", nid)
 				// only check the topic with one replica left
@@ -1661,7 +1661,7 @@ func (self *NsqLookupCoordinator) resetJoinISRState(topicInfo TopicPartitionMeta
 				}
 			}
 			topicInfo.CatchupList = make([]string, 0)
-			for n, _ := range newCatchupList {
+			for n := range newCatchupList {
 				topicInfo.CatchupList = append(topicInfo.CatchupList, n)
 			}
 

internal/clusterinfo/data.go

@@ -9,12 +9,12 @@ import (
 	"strings"
 	"sync"
 
+	"errors"
+	"github.com/blang/semver"
 	"github.com/youzan/nsq/internal/http_api"
 	"github.com/youzan/nsq/internal/stringy"
-	"github.com/blang/semver"
 	"math"
 	"sync/atomic"
-	"errors"
 )
 
 var v1EndpointVersion semver.Version
@@ -92,7 +92,7 @@ func (c *ClusterInfo) GetLookupdTopicsMeta(lookupdHTTPAddrs []string, metaInfo b
 	var errs []error
 
 	type respType struct {
-		Topics []string `json:"topics"`
+		Topics   []string     `json:"topics"`
 		MetaInfo []*TopicInfo `json:"meta_info,omitempty"`
 	}
 
@@ -124,7 +124,7 @@ func (c *ClusterInfo) GetLookupdTopicsMeta(lookupdHTTPAddrs []string, metaInfo b
 			} else {
 				for _, topic := range resp.Topics {
 					topics = append(topics, &TopicInfo{
-						TopicName:topic,
+						TopicName: topic,
 					})
 				}
 			}
@@ -396,15 +396,15 @@ func (c *ClusterInfo) GetLookupdTopicProducers(topic string, lookupdHTTPAddrs []
 }
 
 type TopicInfo struct {
-	TopicName	string	`json:"topic_name"`
-	ExtSupport 	bool 	`json:"extend_support"`
-	Ordered 	bool 	`json:"ordered"`
+	TopicName  string `json:"topic_name"`
+	ExtSupport bool   `json:"extend_support"`
+	Ordered    bool   `json:"ordered"`
 }
 
 type TopicInfoSortByName []*TopicInfo
 
 func (c TopicInfoSortByName) Less(i, j int) bool {
-	return 	c[i].TopicName < c[j].TopicName
+	return c[i].TopicName < c[j].TopicName
 }
 
 func (c TopicInfoSortByName) Swap(i, j int) {
@@ -424,9 +424,9 @@ func (c *ClusterInfo) GetNSQDTopics(nsqdHTTPAddrs []string) ([]*TopicInfo, error
 
 	type respType struct {
 		Topics []struct {
-			Name string `json:"topic_name"`
-			Ordered bool `json:"is_multi_ordered"`
-			ExtSupport bool `json:"is_ext"`
+			Name       string `json:"topic_name"`
+			Ordered    bool   `json:"is_multi_ordered"`
+			ExtSupport bool   `json:"is_ext"`
 		} `json:"topics"`
 	}
 
@@ -450,10 +450,10 @@ func (c *ClusterInfo) GetNSQDTopics(nsqdHTTPAddrs []string) ([]*TopicInfo, error
 			lock.Lock()
 			defer lock.Unlock()
 			for _, topic := range resp.Topics {
-				topics = append(topics,  &TopicInfo{
-					TopicName: topic.Name,
-					Ordered:   topic.Ordered,
-					ExtSupport:topic.ExtSupport,
+				topics = append(topics, &TopicInfo{
+					TopicName:  topic.Name,
+					Ordered:    topic.Ordered,
+					ExtSupport: topic.ExtSupport,
 				})
 			}
 		}(addr)
@@ -464,7 +464,7 @@ func (c *ClusterInfo) GetNSQDTopics(nsqdHTTPAddrs []string) ([]*TopicInfo, error
 		return nil, fmt.Errorf("Failed to query any nsqd: %s", ErrList(errs))
 	}
 
-	sort.Sort(TopicInfoSortByName(topics));
+	sort.Sort(TopicInfoSortByName(topics))
 
 	if len(errs) > 0 {
 		return topics, ErrList(errs)
@@ -921,8 +921,8 @@ func (c *ClusterInfo) GetNSQDStats(producers Producers, selectedTopic string, so
 							StatsdName:     topic.StatsdName,
 							ChannelName:    channel.ChannelName,
 							IsMultiOrdered: topic.IsMultiOrdered,
-							RequeueCount:	channel.RequeueCount,
-							TimeoutCount:	channel.TimeoutCount,
+							RequeueCount:   channel.RequeueCount,
+							TimeoutCount:   channel.TimeoutCount,
 						}
 						channelStatsMap[key] = channelStats
 					}
@@ -1005,7 +1005,7 @@ func (c *ClusterInfo) CreateTopicChannelAfterTopicCreation(topicName string, cha
 	}
 
 	if len(producers) == 0 && len(partitionProducers) == 0 {
-		c.logf(fmt.Sprintf("Producer:%d, PartitionProducers:%d", len(producers), len(partitionProducers)));
+		c.logf(fmt.Sprintf("Producer:%d, PartitionProducers:%d", len(producers), len(partitionProducers)))
 		text := fmt.Sprintf("no producer or partition producer found for Topic:%s, Channel:%s", topicName, channelName)
 		return errors.New(text)
 	}
@@ -1194,7 +1194,7 @@ channelDelete:
 		c.logf("channel %v are not completely deleted", channelName)
 		if retry {
 			//do delete again
-			retry = false;
+			retry = false
 			goto channelDelete
 		} else {
 			c.logf("fail to delete channel %v completely", channelName)