Repository navigation
Expand file tree
/
Copy pathclient.go
More file actions
143 lines (123 loc) · 3.81 KB
/
Copy pathclient.go
File metadata and controls
143 lines (123 loc) · 3.81 KB
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
141
142
143
/*
* Minio Cloud Storage, (C) 2018 Minio, Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package rpc
import (
"context"
"crypto/tls"
"encoding/gob"
"errors"
"fmt"
"net"
"net/http"
"reflect"
"time"
xhttp "github.com/minio/minio/cmd/http"
xnet "github.com/minio/minio/pkg/net"
"golang.org/x/net/http2"
)
// DefaultRPCTimeout - default RPC timeout is one minute.
const DefaultRPCTimeout = 1 * time.Minute
// Client - http based RPC client.
type Client struct {
httpClient *http.Client
httpIdleConnsCloser func()
serviceURL *xnet.URL
}
// Call - calls service method on RPC server.
func (client *Client) Call(serviceMethod string, args, reply interface{}) error {
replyKind := reflect.TypeOf(reply).Kind()
if replyKind != reflect.Ptr {
return fmt.Errorf("rpc reply must be a pointer type, but found %v", replyKind)
}
argBuf := bufPool.Get()
defer bufPool.Put(argBuf)
if err := gobEncodeBuf(args, argBuf); err != nil {
return err
}
callRequest := CallRequest{
Method: serviceMethod,
ArgBytes: argBuf.Bytes(),
}
reqBuf := bufPool.Get()
defer bufPool.Put(reqBuf)
if err := gob.NewEncoder(reqBuf).Encode(callRequest); err != nil {
return err
}
response, err := client.httpClient.Post(client.serviceURL.String(), "", reqBuf)
if err != nil {
return err
}
defer xhttp.DrainBody(response.Body)
if response.StatusCode != http.StatusOK {
return fmt.Errorf("%v rpc call failed with error code %v", serviceMethod, response.StatusCode)
}
var callResponse CallResponse
if err := gob.NewDecoder(response.Body).Decode(&callResponse); err != nil {
return err
}
if callResponse.Error != "" {
return errors.New(callResponse.Error)
}
return gobDecode(callResponse.ReplyBytes, reply)
}
// Close closes all idle connections of the underlying http client
func (client *Client) Close() error {
if client.httpIdleConnsCloser != nil {
client.httpIdleConnsCloser()
}
return nil
}
func newCustomDialContext(timeout time.Duration) func(ctx context.Context, network, addr string) (net.Conn, error) {
return func(ctx context.Context, network, addr string) (net.Conn, error) {
dialer := &net.Dialer{
Timeout: timeout,
KeepAlive: timeout,
DualStack: true,
}
conn, err := dialer.DialContext(ctx, network, addr)
if err != nil {
return nil, err
}
return xhttp.NewTimeoutConn(conn, timeout, timeout), nil
}
}
// NewClient - returns new RPC client.
func NewClient(serviceURL *xnet.URL, tlsConfig *tls.Config, timeout time.Duration) (*Client, error) {
// Transport is exactly same as Go default in https://golang.org/pkg/net/http/#RoundTripper
// except custom DialContext and TLSClientConfig.
tr := &http.Transport{
Proxy: http.ProxyFromEnvironment,
DialContext: newCustomDialContext(timeout),
MaxIdleConnsPerHost: 4096,
MaxIdleConns: 4096,
IdleConnTimeout: 120 * time.Second,
TLSHandshakeTimeout: 30 * time.Second,
ExpectContinueTimeout: 10 * time.Second,
TLSClientConfig: tlsConfig,
DisableCompression: true,
}
if tlsConfig != nil {
// If TLS is enabled configure http2
if err := http2.ConfigureTransport(tr); err != nil {
return nil, err
}
}
return &Client{
httpClient: &http.Client{Transport: tr},
httpIdleConnsCloser: tr.CloseIdleConnections,
serviceURL: serviceURL,
}, nil
}