Check CB Open and init connection for resolve hystrix attempt

Author: Runze Cui

options.go

@@ -25,6 +25,9 @@ type Limiter interface {
 	// Execute allow a customised wrapper around the actual execution.
 	// this is useful for most common circuit breaker implementation
 	Execute(func() error) error
+	// IsCBOpen return whether CB is open
+	// this is useful for implementing additional logic bypass nested hystrix
+	IsCBOpen() bool
 	// ReportResult reports the result of the previously allowed operation.
 	// nil indicates a success, non-nil error usually indicates a failure.
 	ReportResult(result error)
@@ -205,32 +208,38 @@ func (opt *Options) clone() *Options {
 // Scheme is required.
 // There are two connection types: by tcp socket and by unix socket.
 // Tcp connection:
-//		redis://<user>:<password>@<host>:<port>/<db_number>
+//
+//	redis://<user>:<password>@<host>:<port>/<db_number>
+//
 // Unix connection:
-//		unix://<user>:<password>@</path/to/redis.sock>?db=<db_number>
+//
+//	unix://<user>:<password>@</path/to/redis.sock>?db=<db_number>
+//
 // Most Option fields can be set using query parameters, with the following restrictions:
-//	- field names are mapped using snake-case conversion: to set MaxRetries, use max_retries
-//	- only scalar type fields are supported (bool, int, time.Duration)
-//	- for time.Duration fields, values must be a valid input for time.ParseDuration();
-//	  additionally a plain integer as value (i.e. without unit) is intepreted as seconds
-//	- to disable a duration field, use value less than or equal to 0; to use the default
-//	  value, leave the value blank or remove the parameter
-//	- only the last value is interpreted if a parameter is given multiple times
-//	- fields "network", "addr", "username" and "password" can only be set using other
-//	  URL attributes (scheme, host, userinfo, resp.), query paremeters using these
-//	  names will be treated as unknown parameters
-//	- unknown parameter names will result in an error
+//   - field names are mapped using snake-case conversion: to set MaxRetries, use max_retries
+//   - only scalar type fields are supported (bool, int, time.Duration)
+//   - for time.Duration fields, values must be a valid input for time.ParseDuration();
+//     additionally a plain integer as value (i.e. without unit) is intepreted as seconds
+//   - to disable a duration field, use value less than or equal to 0; to use the default
+//     value, leave the value blank or remove the parameter
+//   - only the last value is interpreted if a parameter is given multiple times
+//   - fields "network", "addr", "username" and "password" can only be set using other
+//     URL attributes (scheme, host, userinfo, resp.), query paremeters using these
+//     names will be treated as unknown parameters
+//   - unknown parameter names will result in an error
+//
 // Examples:
-//		redis://user:password@localhost:6789/3?dial_timeout=3&db=1&read_timeout=6s&max_retries=2
-//		is equivalent to:
-//		&Options{
-//			Network:     "tcp",
-//			Addr:        "localhost:6789",
-//			DB:          1,               // path "/3" was overridden by "&db=1"
-//			DialTimeout: 3 * time.Second, // no time unit = seconds
-//			ReadTimeout: 6 * time.Second,
-//			MaxRetries:  2,
-//		}
+//
+//	redis://user:password@localhost:6789/3?dial_timeout=3&db=1&read_timeout=6s&max_retries=2
+//	is equivalent to:
+//	&Options{
+//		Network:     "tcp",
+//		Addr:        "localhost:6789",
+//		DB:          1,               // path "/3" was overridden by "&db=1"
+//		DialTimeout: 3 * time.Second, // no time unit = seconds
+//		ReadTimeout: 6 * time.Second,
+//		MaxRetries:  2,
+//	}
 func ParseURL(redisURL string) (*Options, error) {
 	u, err := url.Parse(redisURL)
 	if err != nil {

redis.go

@@ -265,6 +265,19 @@ func (c *baseClient) withConn(
 		return err
 	}
 
+	// Request is allowed occasionally to close the cb and cb is still in open state
+	// Init the connection to avoid nested hystrix call during connection establishment
+	if limiter.IsCBOpen() {
+		cn, err := c.getConn(ctx)
+		if err != nil {
+			return err
+		}
+
+		defer func() {
+			c.releaseConn(ctx, cn, err)
+		}()
+	}
+
 	err := limiter.Execute(func() error {
 		return c._withConn(ctx, fn)
 	})
@@ -668,26 +681,26 @@ func (c *Client) pubSub() *PubSub {
 // subscription may not be active immediately. To force the connection to wait,
 // you may call the Receive() method on the returned *PubSub like so:
 //
-//    sub := client.Subscribe(queryResp)
-//    iface, err := sub.Receive()
-//    if err != nil {
-//        // handle error
-//    }
+//	sub := client.Subscribe(queryResp)
+//	iface, err := sub.Receive()
+//	if err != nil {
+//	    // handle error
+//	}
 //
-//    // Should be *Subscription, but others are possible if other actions have been
-//    // taken on sub since it was created.
-//    switch iface.(type) {
-//    case *Subscription:
-//        // subscribe succeeded
-//    case *Message:
-//        // received first message
-//    case *Pong:
-//        // pong received
-//    default:
-//        // handle error
-//    }
+//	// Should be *Subscription, but others are possible if other actions have been
+//	// taken on sub since it was created.
+//	switch iface.(type) {
+//	case *Subscription:
+//	    // subscribe succeeded
+//	case *Message:
+//	    // received first message
+//	case *Pong:
+//	    // pong received
+//	default:
+//	    // handle error
+//	}
 //
-//    ch := sub.Channel()
+//	ch := sub.Channel()
 func (c *Client) Subscribe(ctx context.Context, channels ...string) *PubSub {
 	pubsub := c.pubSub()
 	if len(channels) > 0 {