forked from TarsCloud/TarsGo
-
Notifications
You must be signed in to change notification settings - Fork 4
/
adapter.go
140 lines (125 loc) · 2.99 KB
/
adapter.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
package tars
import (
"fmt"
"sync"
"sync/atomic"
"tars/protocol/res/basef"
"time"
"tars/model"
"tars/protocol/res/endpointf"
"tars/protocol/res/requestf"
"tars/transport"
)
// AdapterProxy : Adapter proxy
type AdapterProxy struct {
resp sync.Map
point *endpointf.EndpointF
tarsClient *transport.TarsClient
comm *Communicator
proto model.Protocol
failCount int32
sendCount int32
status bool
closed bool
}
// New : Construct an adapter proxy
func (c *AdapterProxy) New(point *endpointf.EndpointF, comm *Communicator) error {
c.comm = comm
c.point = point
proto := "tcp"
if point.Istcp == 0 {
proto = "udp"
}
conf := &transport.TarsClientConf{
Proto: proto,
//NumConnect: netthread,
QueueLen: ClientQueueLen,
IdleTimeout: ClientIdleTimeout,
ReadTimeout: ClientReadTimeout,
WriteTimeout: ClientWriteTimeout,
}
c.tarsClient = transport.NewTarsClient(fmt.Sprintf("%s:%d", point.Host, point.Port), c, conf)
c.status = true
go c.checkActive()
return nil
}
// ParsePackage : Parse packet from bytes
func (c *AdapterProxy) ParsePackage(buff []byte) (int, int) {
return c.proto.ParsePackage(buff)
}
// Recv : Recover read channel when closed for timeout
func (c *AdapterProxy) Recv(pkg []byte) {
defer func() {
// TODO readCh has a certain probability to be closed after the load, and we need to recover
// Maybe there is a better way
if err := recover(); err != nil {
TLOG.Error("recv pkg painc:", err)
}
}()
packet, err := c.proto.ResponseUnpack(pkg)
if err != nil {
TLOG.Error("decode packet error", err.Error())
return
}
if packet.CPacketType == basef.TARSONEWAY {
return
}
chIF, ok := c.resp.Load(packet.IRequestId)
if ok {
ch := chIF.(chan *requestf.ResponsePacket)
TLOG.Debug("IN:", packet)
ch <- packet
} else {
TLOG.Error("timeout resp,drop it:", packet.IRequestId)
}
}
// Send : Send packet
func (c *AdapterProxy) Send(req *requestf.RequestPacket) error {
TLOG.Debug("send req:", req.IRequestId)
c.sendAdd()
sbuf, err := c.proto.RequestPack(req)
if err != nil {
TLOG.Debug("protocol wrong:", req.IRequestId)
return err
}
return c.tarsClient.Send(sbuf)
}
// GetPoint : Get an endpoint
func (c *AdapterProxy) GetPoint() *endpointf.EndpointF {
return c.point
}
// Close : Close the client
func (c *AdapterProxy) Close() {
c.tarsClient.Close()
c.closed = true
}
func (c *AdapterProxy) sendAdd() {
atomic.AddInt32(&c.sendCount, 1)
}
func (c *AdapterProxy) failAdd() {
atomic.AddInt32(&c.failCount, 1)
}
func (c *AdapterProxy) reset() {
atomic.SwapInt32(&c.sendCount, 0)
atomic.SwapInt32(&c.failCount, 0)
}
func (c *AdapterProxy) checkActive() {
loop := time.NewTicker(AdapterProxyTicker)
count := 0 // Detect if a dead node recovers each minute
for range loop.C {
if c.closed {
loop.Stop()
return
}
if c.failCount > c.sendCount/2 {
c.status = false
}
if !c.status && count > AdapterProxyResetCount {
//TODO USE TAFPING INSTEAD
c.reset()
c.status = true
count = 0
}
count++
}
}