resend on expired endpoint

controller/endpoint/disque.go

@@ -57,7 +57,7 @@ func (conn *DisqueEndpointConn) Send(msg string) error {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 	if conn.ex {
-		return errors.New("expired")
+		return errExpired
 	}
 	conn.t = time.Now()
 	if conn.conn == nil {

controller/endpoint/endpoint.go

@@ -9,6 +9,8 @@ import (
 	"time"
 )
 
+var errExpired = errors.New("expired")
+
 // EndpointProtocol is the type of protocol that the endpoint represents.
 type EndpointProtocol string
 
@@ -82,6 +84,7 @@ func (epc *EndpointManager) Validate(url string) error {
 }
 
 func (epc *EndpointManager) Send(endpoint, val string) error {
+	for {
 		epc.mu.Lock()
 		conn, ok := epc.conns[endpoint]
 		if !ok || conn.Expired() {
@@ -105,7 +108,18 @@ func (epc *EndpointManager) Send(endpoint, val string) error {
 			epc.conns[endpoint] = conn
 		}
 		epc.mu.Unlock()
-	return conn.Send(val)
+		err := conn.Send(val)
+		if err != nil {
+			if err == errExpired {
+				// it's possible that the connection has expired in-between
+				// the last conn.Expired() check and now. If so, we should
+				// just try the send again.
+				continue
+			}
+			return err
+		}
+		return nil
+	}
 }
 
 func parseEndpoint(s string) (Endpoint, error) {

controller/endpoint/grpc.go

@@ -57,7 +57,7 @@ func (conn *GRPCEndpointConn) Send(msg string) error {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 	if conn.ex {
-		return errors.New("expired")
+		return errExpired
 	}
 	conn.t = time.Now()
 	if conn.conn == nil {

controller/endpoint/http.go

@@ -2,7 +2,6 @@ package endpoint
 
 import (
 	"bytes"
-	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -48,7 +47,7 @@ func (conn *HTTPEndpointConn) Send(msg string) error {
 	conn.mu.Lock()
 	defer conn.mu.Unlock()
 	if conn.ex {
-		return errors.New("expired")
+		return errExpired
 	}
 	conn.t = time.Now()
 	if conn.client == nil {

controller/endpoint/redis.go

@@ -56,7 +56,7 @@ func (conn *RedisEndpointConn) Send(msg string) error {
 	defer conn.mu.Unlock()
 
 	if conn.ex {
-		return errors.New("expired")
+		return errExpired
 	}
 
 	conn.t = time.Now()