Merge pull request #209 from mreiferson/retry-lookup

consumer: retry nsqlookupd queries

api_request.go

@@ -27,6 +27,7 @@ func (c *deadlinedConn) Write(b []byte) (n int, err error) {
 
 func newDeadlineTransport(timeout time.Duration) *http.Transport {
  transport := &http.Transport{
+ DisableKeepAlives: true,
  Dial: func(netw, addr string) (net.Conn, error) {
  c, err := net.DialTimeout(netw, addr, timeout)
  if err != nil {

conn.go

@@ -487,7 +487,6 @@ func (c *Conn) readLoop() {
  }
 
  frameType, data, err := ReadUnpackedResponse(c)
-
  if err != nil {
  if err == io.EOF && atomic.LoadInt32(&c.closeFlag) == 1 {
  goto exit

consumer.go

@@ -454,6 +454,9 @@ type peerInfo struct {
 //
 // initiate a connection to any new producers that are identified.
 func (r *Consumer) queryLookupd() {
+ retries := 0
+
+retry:
  endpoint := r.nextLookupdEndpoint()
 
  r.log(LogLevelInfo, "querying nsqlookupd %s", endpoint)
@@ -462,6 +465,11 @@ func (r *Consumer) queryLookupd() {
  err := apiRequestNegotiateV1("GET", endpoint, nil, &data)
  if err != nil {
  r.log(LogLevelError, "error querying nsqlookupd (%s) - %s", endpoint, err)
+ retries++
+ if retries < 3 {
+ r.log(LogLevelInfo, "retrying with next nsqlookupd")
+ goto retry
+ }
  return
  }
 

consumer_test.go

@@ -64,6 +64,9 @@ func SendMessage(t *testing.T, port int, topic string, method string, body []byt
  t.Fatalf(err.Error())
  return
  }
+ if resp.StatusCode != 200 {
+ t.Fatalf("%s status code: %d", method, resp.StatusCode)
+ }
  resp.Body.Close()
 }
 
@@ -165,17 +168,17 @@ func consumerTest(t *testing.T, cb func(c *Config)) {
  }
  topicName = topicName + strconv.Itoa(int(time.Now().Unix()))
  q, _ := NewConsumer(topicName, "ch", config)
- // q.SetLogger(nullLogger, LogLevelInfo)
+ q.SetLogger(log.New(os.Stderr, "", log.Flags()), LogLevelDebug)
 
  h := &MyTestHandler{
  t: t,
  q: q,
  }
  q.AddHandler(h)
 
- SendMessage(t, 4151, topicName, "put", []byte(`{"msg":"single"}`))
- SendMessage(t, 4151, topicName, "mput", []byte("{\"msg\":\"double\"}\n{\"msg\":\"double\"}"))
- SendMessage(t, 4151, topicName, "put", []byte("TOBEFAILED"))
+ SendMessage(t, 4151, topicName, "pub", []byte(`{"msg":"single"}`))
+ SendMessage(t, 4151, topicName, "mpub", []byte("{\"msg\":\"double\"}\n{\"msg\":\"double\"}"))
+ SendMessage(t, 4151, topicName, "pub", []byte("TOBEFAILED"))
  h.messagesSent = 4
 
  addr := "127.0.0.1:4150"