-
-
Notifications
You must be signed in to change notification settings - Fork 180
/
node.go
133 lines (107 loc) · 2.74 KB
/
node.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
package gorethink
import (
"sync"
p "gopkg.in/gorethink/gorethink.v2/ql2"
)
// Node represents a database server in the cluster
type Node struct {
ID string
Host Host
aliases []Host
cluster *Cluster
pool *Pool
mu sync.RWMutex
closed bool
}
func newNode(id string, aliases []Host, cluster *Cluster, pool *Pool) *Node {
node := &Node{
ID: id,
Host: aliases[0],
aliases: aliases,
cluster: cluster,
pool: pool,
}
return node
}
// Closed returns true if the node is closed
func (n *Node) Closed() bool {
n.mu.RLock()
defer n.mu.RUnlock()
return n.closed
}
// Close closes the session
func (n *Node) Close(optArgs ...CloseOpts) error {
n.mu.Lock()
defer n.mu.Unlock()
if n.closed {
return nil
}
if len(optArgs) >= 1 {
if optArgs[0].NoReplyWait {
n.NoReplyWait()
}
}
if n.pool != nil {
n.pool.Close()
}
n.pool = nil
n.closed = true
return nil
}
// SetInitialPoolCap sets the initial capacity of the connection pool.
func (n *Node) SetInitialPoolCap(idleConns int) {
n.pool.SetInitialPoolCap(idleConns)
}
// SetMaxIdleConns sets the maximum number of connections in the idle
// connection pool.
func (n *Node) SetMaxIdleConns(idleConns int) {
n.pool.SetMaxIdleConns(idleConns)
}
// SetMaxOpenConns sets the maximum number of open connections to the database.
func (n *Node) SetMaxOpenConns(openConns int) {
n.pool.SetMaxOpenConns(openConns)
}
// NoReplyWait ensures that previous queries with the noreply flag have been
// processed by the server. Note that this guarantee only applies to queries
// run on the given connection
func (n *Node) NoReplyWait() error {
return n.pool.Exec(Query{
Type: p.Query_NOREPLY_WAIT,
})
}
// Query executes a ReQL query using this nodes connection pool.
func (n *Node) Query(q Query) (cursor *Cursor, err error) {
if n.Closed() {
return nil, ErrInvalidNode
}
return n.pool.Query(q)
}
// Exec executes a ReQL query using this nodes connection pool.
func (n *Node) Exec(q Query) (err error) {
if n.Closed() {
return ErrInvalidNode
}
return n.pool.Exec(q)
}
// Server returns the server name and server UUID being used by a connection.
func (n *Node) Server() (ServerResponse, error) {
var response ServerResponse
if n.Closed() {
return response, ErrInvalidNode
}
return n.pool.Server()
}
type nodeStatus struct {
ID string `gorethink:"id"`
Name string `gorethink:"name"`
Status string `gorethink:"status"`
Network struct {
Hostname string `gorethink:"hostname"`
ClusterPort int64 `gorethink:"cluster_port"`
ReqlPort int64 `gorethink:"reql_port"`
CanonicalAddresses []struct {
Host string `gorethink:"host"`
Port int64 `gorethink:"port"`
} `gorethink:"canonical_addresses"`
} `gorethink:"network"`
}