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
|
package rethinkdb
import (
"sync"
"golang.org/x/net/context"
p "gopkg.in/rethinkdb/rethinkdb-go.v6/ql2"
)
// Node represents a database server in the cluster
type Node struct {
ID string
Host Host
aliases []Host
pool *Pool
mu sync.RWMutex
closed bool
}
func newNode(id string, aliases []Host, pool *Pool) *Node {
node := &Node{
ID: id,
Host: aliases[0],
aliases: aliases,
pool: pool,
}
return node
}
// Closed returns true if the node is connClosed
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(nil, Query{ // nil = connection opts' timeout
Type: p.Query_NOREPLY_WAIT,
})
}
// Query executes a ReQL query using this nodes connection pool.
func (n *Node) Query(ctx context.Context, q Query) (cursor *Cursor, err error) {
if n.Closed() {
return nil, ErrInvalidNode
}
return n.pool.Query(ctx, q)
}
// Exec executes a ReQL query using this nodes connection pool.
func (n *Node) Exec(ctx context.Context, q Query) (err error) {
if n.Closed() {
return ErrInvalidNode
}
return n.pool.Exec(ctx, 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 `rethinkdb:"id"`
Name string `rethinkdb:"name"`
Network nodeStatusNetwork `rethinkdb:"network"`
}
type nodeStatusNetwork struct {
Hostname string `rethinkdb:"hostname"`
ClusterPort int64 `rethinkdb:"cluster_port"`
ReqlPort int64 `rethinkdb:"reql_port"`
CanonicalAddresses []nodeStatusNetworkAddr `rethinkdb:"canonical_addresses"`
}
type nodeStatusNetworkAddr struct {
Host string `rethinkdb:"host"`
Port int64 `rethinkdb:"port"`
}
|
