finish advanced circuit breaker

resiliency/06_circuit_breaker/breaker_options.go

@@ -14,7 +14,7 @@ import (
 )
 
 //BreakConditionWatcher check state
-type BreakConditionWatcher func(cnter counters) bool
+type BreakConditionWatcher func(state State, cnter counters) bool
 
 //StateChangedEventHandler set event handle
 type StateChangedEventHandler func(name string, from State, to State)
@@ -24,18 +24,19 @@ type Option func(opts *Options)
 
 //Options for breaker
 type Options struct {
-	Name              string
-	Expiry            time.Time
-	Interval, Timeout time.Duration
-	MaxRequests       uint32
-	WhenToBreak       BreakConditionWatcher //是否应该断开电路(打开电路开关)
-	OnStateChanged    StateChangedEventHandler
-	ShoulderToOpen    uint32
-	Ctx               context.Context
+	Name               string
+	Expiry             time.Time
+	Interval, Timeout  time.Duration
+	MaxRequests        uint32
+	CanOpen            BreakConditionWatcher //是否应该断开电路(打开电路开关)
+	CanClose           BreakConditionWatcher //if we should close switch
+	OnStateChanged     StateChangedEventHandler
+	ShoulderHalfToOpen uint32
+	Ctx                context.Context
 }
 
-//Name of breaker
-func Name(name string) Option {
+//ActionName of breaker
+func ActionName(name string) Option {
 	return func(opts *Options) {
 		opts.Name = name
 	}
@@ -66,6 +67,13 @@ func MaxRequests(maxRequests uint32) Option {
 	}
 }
 
+//WithShoulderHalfToOpen of breaker
+func WithShoulderHalfToOpen(shoulderHalfToOpen uint32) Option {
+	return func(opts *Options) {
+		opts.ShoulderHalfToOpen = shoulderHalfToOpen
+	}
+}
+
 //Expiry of breaker
 func Expiry(expiry time.Time) Option {
 	return func(opts *Options) {
@@ -80,9 +88,16 @@ func WithStateChanged(handler StateChangedEventHandler) Option {
 	}
 }
 
-//BreakIf check traffic state ,to see if request can go
-func BreakIf(whenCondition BreakConditionWatcher) Option {
+//WithBreakCondition check traffic state ,to see if request can go
+func WithBreakCondition(whenCondition BreakConditionWatcher) Option {
 	return func(opts *Options) {
-		opts.WhenToBreak = whenCondition
+		opts.CanOpen = whenCondition
+	}
+}
+
+//WithCloseCondition check traffic state ,to see if request can go
+func WithCloseCondition(whenCondition BreakConditionWatcher) Option {
+	return func(opts *Options) {
+		opts.CanClose = whenCondition
 	}
 }

resiliency/06_circuit_breaker/circuit_breaker_adv.go

@@ -42,11 +42,11 @@ const (
 
 //RequestBreaker for protection
 type RequestBreaker struct {
-	options    Options
-	mutex      sync.Mutex
-	state      State
-	generation uint64
-	cnter      counters
+	options  Options
+	mutex    sync.Mutex
+	state    State
+	cnter    counters
+	preState State
 }
 
 //NewRequestBreaker return a breaker
@@ -55,10 +55,11 @@ func NewRequestBreaker(opts ...Option) *RequestBreaker {
 	defaultOptions := Options{
 		Name:           "defaultBreakerName",
 		Expiry:         time.Now().Add(time.Second * 20),
-		Interval:       time.Second * 2,  // interval to check status
-		Timeout:        time.Second * 60, //default to 60 seconds
+		Interval:       time.Second * 10, // interval to check  closed status,default 10 seconds
+		Timeout:        time.Second * 60, //timeout to check open, default 60 seconds
 		MaxRequests:    5,
-		WhenToBreak:    func(cnter counters) bool { return cnter.ConsecutiveFailures > 2 },
+		CanOpen:        func(current State, cnter counters) bool { return cnter.ConsecutiveFailures > 2 },
+		CanClose:       func(current State, cnter counters) bool { return cnter.ConsecutiveSuccesses > 2 },
 		OnStateChanged: func(name string, fromPre State, toCurrent State) {},
 	}
 
@@ -68,74 +69,102 @@ func NewRequestBreaker(opts ...Option) *RequestBreaker {
 	}
 
 	return &RequestBreaker{
-		options:    defaultOptions,
-		cnter:      counters{},
-		generation: 0,
+		options:  defaultOptions,
+		cnter:    counters{},
+		state:    StateUnknown,
+		preState: StateUnknown,
 	}
 }
 
+func (rb *RequestBreaker) changeStateTo(state State) {
+	rb.preState = rb.state
+	rb.state = state
+	rb.cnter.Reset()
+}
+
+func (rb *RequestBreaker) beforeRequest() error {
+
+	rb.mutex.Lock()
+	defer rb.mutex.Unlock()
+	fmt.Println("before do request:", rb.cnter.Total())
+
+	switch rb.state {
+	case StateOpen:
+		//如果是断开状态，并且超时了，转到半开状态，记录
+		if rb.options.Expiry.Before(time.Now()) {
+			rb.changeStateTo(StateHalfOpen)
+			rb.options.Expiry = time.Now().Add(rb.options.Timeout)
+			return nil
+		}
+	case StateClosed:
+		if rb.options.Expiry.Before(time.Now()) {
+			rb.cnter.Reset()
+			rb.options.Expiry = time.Now().Add(rb.options.Interval)
+		}
+
+	}
+
+	return ErrTooManyRequests
+
+}
+
 // Do the given requested work if the RequestBreaker accepts it.
 // Do returns an error instantly if the RequestBreaker rejects the request.
 // Otherwise, Execute returns the result of the request.
 // If a panic occurs in the request, the RequestBreaker handles it as an error and causes the same panic again.
 func (rb *RequestBreaker) Do(work func(ctx context.Context) (interface{}, error)) (interface{}, error) {
 
-	preState := StateUnknown
-
 	//before
-	fmt.Println("before do : request:", rb.cnter.Total())
-	rb.mutex.Lock()
 
-	//如果是断开状态，并且超时了，转到半开状态，记录
-	if rb.state == StateOpen && rb.options.Expiry.Before(time.Now()) {
-		preState = rb.state
-		rb.state = StateHalfOpen
-		rb.cnter.Reset()
-	} else {
-		return nil, ErrTooManyRequests
+	if err := rb.beforeRequest(); err != nil {
+		return nil, err
 	}
 
-	rb.mutex.Unlock()
-
 	//do work
 	//do work from requested user
 	result, err := work(rb.options.Ctx)
 
+	//after work
+	rb.afterRequest(err)
+
+	return result, err
+}
+
+func (rb *RequestBreaker) afterRequest(resultErr error) {
+
 	rb.mutex.Lock()
-	//失败了
-	if err != nil {
-		rb.cnter.Count(FailureState)
-		//如果是在半开状态下的失败，立即打开开关
-		if rb.state == StateHalfOpen {
-			rb.state = StateOpen                                                 //转为打开
-			rb.options.OnStateChanged(rb.options.Name, StateHalfOpen, StateOpen) //半开到打开
-		} else if rb.state == StateClosed {
-			if rb.options.WhenToBreak(rb.cnter) {
-				rb.state = StateOpen                                               //打开开关
-				rb.options.OnStateChanged(rb.options.Name, StateClosed, StateOpen) //关闭到打开
-				rb.cnter.Reset()
+	defer rb.mutex.Unlock()
+	//after
+	fmt.Println("after do request:", rb.cnter.Total())
+
+	if resultErr != nil {
+		//失败了,handle 失败
+		rb.cnter.Count(FailureState, rb.preState == rb.state)
+		switch rb.state {
+		case StateHalfOpen, StateClosed:
+			if rb.options.CanOpen(rb.state, rb.cnter) {
+				rb.changeStateTo(StateOpen)                                     //打开开关
+				rb.options.OnStateChanged(rb.options.Name, rb.state, StateOpen) //关闭到打开
 			}
 		}
-
 	} else {
-		//成功了.
-		rb.cnter.Count(SuccessState)
-		if rb.state == StateHalfOpen {
-			if preState == StateOpen {
-				preState = StateHalfOpen
+		//success !
+		rb.cnter.Count(SuccessState, rb.preState == rb.state)
+
+		switch rb.state {
+		case StateHalfOpen:
+			if rb.preState == StateOpen {
+				//	rb.changeStateTo(StateHalfOpen) //already handled in beforeRequest,Only fire StateChange Event
 				rb.options.OnStateChanged(rb.options.Name, StateOpen, StateHalfOpen) //打开到半开
 			}
-			if rb.cnter.ConsecutiveSuccesses >= rb.options.ShoulderToOpen {
-				rb.state = StateClosed
+			if rb.cnter.ConsecutiveSuccesses >= rb.options.ShoulderHalfToOpen {
+				rb.changeStateTo(StateClosed)
+				rb.options.Expiry = time.Now().Add(rb.options.Interval)
 				rb.options.OnStateChanged(rb.options.Name, StateHalfOpen, StateClosed) //半开到关闭
 			}
 
 		}
 
 	}
-	rb.mutex.Unlock()
-	//after
-	fmt.Println("after do : request:", rb.cnter.Total())
 
-	return result, err
 }

resiliency/06_circuit_breaker/circuit_breaker_test.go

@@ -23,12 +23,27 @@ var onStateChangeEvent = func(name string, from, to State) {
 	fmt.Println("name:", name, "from:", from, "to", to)
 }
 
-var whenConditionOccurred = func(cnter counters) bool {
+var canOpenSwitch = func(current State, cnter counters) bool {
+
+	if current == StateHalfOpen {
+		return cnter.ConsecutiveFailures > 2
+	}
+
 	//失败率，可以由用户自己定义
 	failureRatio := float64(cnter.TotalFailures) / float64(cnter.Requests)
 	return cnter.Requests >= 3 && failureRatio >= 0.6
 }
 
+var canCloseSwitch = func(current State, cnter counters) bool {
+	//失败率，可以由用户自己定义
+	if cnter.ConsecutiveSuccesses > 2 {
+		return true
+	}
+	//
+	successRatio := float64(cnter.TotalFailures) / float64(cnter.Requests)
+	return cnter.Requests >= 3 && successRatio >= 0.6
+}
+
 func TestObjectBreaker(t *testing.T) {
 
 	jobToDo := func(ctx context.Context) (interface{}, error) {
@@ -44,7 +59,11 @@ func TestObjectBreaker(t *testing.T) {
 		return body, nil
 	}
 
-	breaker = NewRequestBreaker(Name("HTTP GET"), BreakIf(whenConditionOccurred), WithStateChanged(onStateChangeEvent))
+	breaker = NewRequestBreaker(ActionName("HTTP GET"),
+		WithBreakCondition(canOpenSwitch),
+		WithCloseCondition(canCloseSwitch),
+		WithShoulderHalfToOpen(2),
+		WithStateChanged(onStateChangeEvent))
 
 	body, err := breaker.Do(jobToDo)
 	if err != nil {

resiliency/06_circuit_breaker/circuit_counter.go

@@ -33,7 +33,7 @@ type OperationState int
 
 //ICounter interface
 type ICounter interface {
-	Count(OperationState)
+	Count(OperationState, bool)
 	LastActivity() time.Time
 	Reset()
 	Total() uint32
@@ -63,13 +63,19 @@ func (c *counters) Reset() {
 }
 
 //Count the failure and success
-func (c *counters) Count(statue OperationState) {
+func (c *counters) Count(statue OperationState, isConsecutive bool) {
 
 	switch statue {
 	case FailureState:
-		c.ConsecutiveFailures++
+		c.TotalFailures++
+		if isConsecutive || c.ConsecutiveFailures == 0 {
+			c.ConsecutiveFailures++
+		}
 	case SuccessState:
-		c.ConsecutiveSuccesses++
+		c.TotalSuccesses++
+		if isConsecutive || c.ConsecutiveSuccesses == 0 {
+			c.ConsecutiveSuccesses++
+		}
 	}
 	c.Requests++
 	c.lastActivity = time.Now() //更新活动时间