nsqd: SUB: fix ephemeral channel race condition with reconnecting cli…

…ent.

It's possible for a client reconnecting quickly and subscribed to an ephemeral channel to race with nsqd's cleanup of said ephemeral channel, as documented in nsqio/go-nsq#206.

Fixes nsqio#883

nsqd/protocol_v2.go

@@ -615,9 +615,20 @@ func (p *protocolV2) SUB(client *clientV2, params [][]byte) ([]byte, error) {
 		return nil, err
 	}
 
-	topic := p.ctx.nsqd.GetTopic(topicName)
-	channel := topic.GetChannel(channelName)
-	channel.AddClient(client.ID, client)
+	for {
+		topic := p.ctx.nsqd.GetTopic(topicName)
+		if topic.Exiting() {
+			time.Sleep(1 * time.Microsecond)
+			continue
+		}
+		channel := topic.GetChannel(channelName)
+		if channel.Exiting() {
+			time.Sleep(1 * time.Microsecond)
+			continue
+		}
+		channel.AddClient(client.ID, client)
+		break
+	}
 
 	atomic.StoreInt32(&client.State, stateSubscribed)
 	client.Channel = channel