nsqd: cover additional edge cases

Author: jehiah

nsqd/http.go

@@ -180,8 +180,12 @@ func (s *httpServer) getTopicFromQuery(req *http.Request) (url.Values, *Topic, e
 	if !protocol.IsValidTopicName(topicName) {
 		return nil, nil, http_api.Err{400, "INVALID_TOPIC"}
 	}
+	topic := s.ctx.nsqd.GetTopic(topicName)
+	if topic == nil {
+		return nil, nil, http_api.Err{503, "EXITING"}
+	}
 
-	return reqParams, s.nsqd.GetTopic(topicName), nil
+	return reqParams, topic, nil
 }
 
 func (s *httpServer) doPUB(w http.ResponseWriter, req *http.Request, ps httprouter.Params) (interface{}, error) {
@@ -431,7 +435,10 @@ func (s *httpServer) doCreateChannel(w http.ResponseWriter, req *http.Request, p
 	if err != nil {
 		return nil, err
 	}
-	topic.GetChannel(channelName)
+	ch := topic.GetChannel(channelName)
+	if ch == nil {
+		return nil, http_api.Err{503, "EXITING"}
+	}
 	return nil, nil
 }
 

nsqd/nsqd.go

@@ -478,8 +478,21 @@ func (n *NSQD) GetTopic(topicName string) *Topic {
 		n.Unlock()
 		return t
 	}
+	if atomic.LoadInt32(&t.isDraining) == 1 {
+		// don't create new topics when nsqd is draining
+		return nil
+	}
+
 	deleteCallback := func(t *Topic) {
 		n.DeleteExistingTopic(t.name)
+		if atomic.LoadInt32(&t.isDraining) == 1 {
+			n.RLock()
+			topicCount := len(n.topicMap)
+			n.RUnlock()
+			if topicCount == 0 {
+				n.Exit()
+			}
+		}
 	}
 	t = NewTopic(topicName, n, deleteCallback)
 	n.topicMap[topicName] = t
@@ -563,6 +576,9 @@ func (n *NSQD) StartDraining() error {
 	for _, t := range n.topicMap {
 		t.StartDraining()
 	}
+	if len(n.topicMap) == 0 {
+		n.Exit()
+	}
 	return nil
 }