/
rpc_writer.go
83 lines (68 loc) · 1.79 KB
/
rpc_writer.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
package tunnel
import (
"net"
"net/rpc"
"mongoshake/common"
LOG "github.com/vinllen/log4go"
)
type RPCWriter struct {
RemoteAddr string
// for golang rpc
tcpAddr *net.TCPAddr
rpcConn *net.TCPConn
rpcClient *rpc.Client
}
func (tunnel *RPCWriter) Send(message *WMessage) int64 {
var err error
if tunnel.rpcConn == nil {
// we try just one time as higher layer will handle this error
if tunnel.rpcConn, err = net.DialTCP("tcp", nil, tunnel.tcpAddr); err != nil {
LOG.Critical("Remote rpc server connect failed. %v", err)
utils.YieldInMs(3000)
tunnel.rpcConn = nil
return ReplyNetworkOpFail
}
tunnel.rpcClient = rpc.NewClient(tunnel.rpcConn)
}
message.Tag |= MsgResident
// DON'T need to check len(logs) == 0. It may be a reasonable
// probe request sending
var reply int64
err = tunnel.rpcClient.Call("TunnelRPC.Transfer", message.TMessage, &reply)
if err != nil {
LOG.Error("Remote rpc server send error[%v]", err)
// error is from network or rpc system.
tunnel.rpcClient.Close()
tunnel.rpcConn.Close()
tunnel.rpcConn = nil
return ReplyError
}
return reply
}
func (tunnel *RPCWriter) Prepare() bool {
var address *net.TCPAddr
var conn *net.TCPConn
var err error
if address, err = net.ResolveTCPAddr("tcp", tunnel.RemoteAddr); err != nil {
LOG.Critical("Resolve rpc server address failed. %v", err)
return false
}
tunnel.tcpAddr = address
// check connection on initial stage
if !InitialStageChecking {
return true
}
if conn, err = net.DialTCP("tcp", nil, address); err != nil {
LOG.Critical("Remote rpc server connect failed. %v", err)
return false
}
// just test the connection
conn.Close()
return true
}
func (tunnel *RPCWriter) AckRequired() bool {
return true
}
func (tunnel *RPCWriter) ParsedLogsRequired() bool {
return false
}