improve http handler

pkg/chain/node.go

@@ -1,15 +1,22 @@
 package chain
 
+import (
+	"sync"
+	"time"
+)
+
 type Node struct {
 	name      string
 	addr      string
 	transport *Transport
+	marker    *failMarker
 }
 
 func NewNode(name, addr string) *Node {
 	return &Node{
-		name: name,
-		addr: addr,
+		name:   name,
+		addr:   addr,
+		marker: &failMarker{},
 	}
 }
 
@@ -45,15 +52,72 @@ func (g *NodeGroup) AddNode(node *Node) {
 	g.nodes = append(g.nodes, node)
 }
 
-func (g *NodeGroup) WithSelector(selector Selector) {
+func (g *NodeGroup) WithSelector(selector Selector) *NodeGroup {
 	g.selector = selector
+	return g
 }
 
 func (g *NodeGroup) Next() *Node {
+	if g == nil || len(g.nodes) == 0 {
+		return nil
+	}
+
 	selector := g.selector
 	if selector == nil {
-		// selector = defaultSelector
 		return g.nodes[0]
 	}
+
 	return selector.Select(g.nodes...)
 }
+
+type failMarker struct {
+	failTime  int64
+	failCount uint32
+	mux       sync.RWMutex
+}
+
+func (m *failMarker) FailTime() int64 {
+	if m == nil {
+		return 0
+	}
+
+	m.mux.RLock()
+	defer m.mux.RUnlock()
+
+	return m.failTime
+}
+
+func (m *failMarker) FailCount() uint32 {
+	if m == nil {
+		return 0
+	}
+
+	m.mux.RLock()
+	defer m.mux.RUnlock()
+
+	return m.failCount
+}
+
+func (m *failMarker) Mark() {
+	if m == nil {
+		return
+	}
+
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	m.failTime = time.Now().Unix()
+	m.failCount++
+}
+
+func (m *failMarker) Reset() {
+	if m == nil {
+		return
+	}
+
+	m.mux.Lock()
+	defer m.mux.Unlock()
+
+	m.failTime = 0
+	m.failCount = 0
+}

pkg/chain/route.go

@@ -22,26 +22,34 @@ func (r *Route) Connect(ctx context.Context) (conn net.Conn, err error) {
 	node := r.nodes[0]
 	cc, err := node.Transport().Dial(ctx, r.nodes[0].Addr())
 	if err != nil {
+		node.marker.Mark()
 		return
 	}
 
 	cn, err := node.Transport().Handshake(ctx, cc)
 	if err != nil {
 		cc.Close()
+		node.marker.Mark()
 		return
 	}
+	node.marker.Reset()
 
 	preNode := node
 	for _, node := range r.nodes[1:] {
 		cc, err = preNode.Transport().Connect(ctx, cn, "tcp", node.Addr())
 		if err != nil {
 			cn.Close()
+			node.marker.Mark()
 			return
 		}
 		cc, err = node.transport.Handshake(ctx, cc)
 		if err != nil {
 			cn.Close()
+			node.marker.Mark()
+			return
 		}
+		node.marker.Reset()
+
 		cn = cc
 		preNode = node
 	}

pkg/chain/selector.go

@@ -1,19 +1,24 @@
 package chain
 
+import (
+	"math/rand"
+	"net"
+	"strconv"
+	"sync"
+	"sync/atomic"
+	"time"
+)
+
+// default options for FailFilter
+const (
+	DefaultMaxFails    = 1
+	DefaultFailTimeout = 30 * time.Second
+)
+
 var (
 	defaultSelector Selector = NewSelector(nil)
 )
 
-type Filter interface {
-	Filter(nodes ...*Node) []*Node
-	String() string
-}
-
-type Strategy interface {
-	Apply(nodes ...*Node) *Node
-	String() string
-}
-
 type Selector interface {
 	Select(nodes ...*Node) *Node
 }
@@ -39,3 +44,115 @@ func (s *selector) Select(nodes ...*Node) *Node {
 	}
 	return s.strategy.Apply(nodes...)
 }
+
+type Strategy interface {
+	Apply(nodes ...*Node) *Node
+}
+
+// RoundStrategy is a strategy for node selector.
+// The node will be selected by round-robin algorithm.
+type RoundRobinStrategy struct {
+	counter uint64
+}
+
+func (s *RoundRobinStrategy) Apply(nodes ...*Node) *Node {
+	if len(nodes) == 0 {
+		return nil
+	}
+
+	n := atomic.AddUint64(&s.counter, 1) - 1
+	return nodes[int(n%uint64(len(nodes)))]
+}
+
+// RandomStrategy is a strategy for node selector.
+// The node will be selected randomly.
+type RandomStrategy struct {
+	Seed int64
+	rand *rand.Rand
+	once sync.Once
+	mux  sync.Mutex
+}
+
+func (s *RandomStrategy) Apply(nodes ...*Node) *Node {
+	s.once.Do(func() {
+		seed := s.Seed
+		if seed == 0 {
+			seed = time.Now().UnixNano()
+		}
+		s.rand = rand.New(rand.NewSource(seed))
+	})
+	if len(nodes) == 0 {
+		return nil
+	}
+
+	s.mux.Lock()
+	defer s.mux.Unlock()
+
+	r := s.rand.Int()
+
+	return nodes[r%len(nodes)]
+}
+
+// FIFOStrategy is a strategy for node selector.
+// The node will be selected from first to last,
+// and will stick to the selected node until it is failed.
+type FIFOStrategy struct{}
+
+// Apply applies the fifo strategy for the nodes.
+func (s *FIFOStrategy) Apply(nodes ...*Node) *Node {
+	if len(nodes) == 0 {
+		return nil
+	}
+	return nodes[0]
+}
+
+type Filter interface {
+	Filter(nodes ...*Node) []*Node
+}
+
+// FailFilter filters the dead node.
+// A node is marked as dead if its failed count is greater than MaxFails.
+type FailFilter struct {
+	MaxFails    int
+	FailTimeout time.Duration
+}
+
+// Filter filters dead nodes.
+func (f *FailFilter) Filter(nodes ...*Node) []*Node {
+	maxFails := f.MaxFails
+	if maxFails == 0 {
+		maxFails = DefaultMaxFails
+	}
+	failTimeout := f.FailTimeout
+	if failTimeout == 0 {
+		failTimeout = DefaultFailTimeout
+	}
+
+	if len(nodes) <= 1 || maxFails < 0 {
+		return nodes
+	}
+	var nl []*Node
+	for _, node := range nodes {
+		if node.marker.FailCount() < uint32(maxFails) ||
+			time.Since(time.Unix(node.marker.FailTime(), 0)) >= failTimeout {
+			nl = append(nl, node)
+		}
+	}
+	return nl
+}
+
+// InvalidFilter filters the invalid node.
+// A node is invalid if its port is invalid (negative or zero value).
+type InvalidFilter struct{}
+
+// Filter filters invalid nodes.
+func (f *InvalidFilter) Filter(nodes ...*Node) []*Node {
+	var nl []*Node
+	for _, node := range nodes {
+		_, sport, _ := net.SplitHostPort(node.Addr())
+		if port, _ := strconv.Atoi(sport); port > 0 {
+			nl = append(nl, node)
+		}
+	}
+	return nl
+}