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
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
|
// Copyright 2009 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
package rpc
import (
"bufio";
"gob";
"http";
"io";
"log";
"net";
"os";
"strconv";
"sync";
)
// Call represents an active RPC.
type Call struct {
ServiceMethod string; // The name of the service and method to call.
Args interface{}; // The argument to the function (*struct).
Reply interface{}; // The reply from the function (*struct).
Error os.Error; // After completion, the error status.
Done chan *Call; // Strobes when call is complete; value is the error status.
seq uint64;
}
// Client represents an RPC Client.
// There may be multiple outstanding Calls associated
// with a single Client.
type Client struct {
mutex sync.Mutex; // protects pending, seq
shutdown os.Error; // non-nil if the client is shut down
sending sync.Mutex;
seq uint64;
conn io.ReadWriteCloser;
enc *gob.Encoder;
dec *gob.Decoder;
pending map[uint64] *Call;
}
func (client *Client) send(c *Call) {
// Register this call.
client.mutex.Lock();
if client.shutdown != nil {
c.Error = client.shutdown;
client.mutex.Unlock();
_ = c.Done <- c; // do not block
return;
}
c.seq = client.seq;
client.seq++;
client.pending[c.seq] = c;
client.mutex.Unlock();
// Encode and send the request.
request := new(Request);
client.sending.Lock();
request.Seq = c.seq;
request.ServiceMethod = c.ServiceMethod;
client.enc.Encode(request);
err := client.enc.Encode(c.Args);
if err != nil {
panicln("rpc: client encode error:", err);
}
client.sending.Unlock();
}
func (client *Client) input() {
var err os.Error;
for err == nil {
response := new(Response);
err = client.dec.Decode(response);
if err != nil {
if err == os.EOF {
err = io.ErrUnexpectedEOF;
}
break
}
seq := response.Seq;
client.mutex.Lock();
c := client.pending[seq];
client.pending[seq] = c, false;
client.mutex.Unlock();
err = client.dec.Decode(c.Reply);
c.Error = os.ErrorString(response.Error);
// We don't want to block here. It is the caller's responsibility to make
// sure the channel has enough buffer space. See comment in Go().
_ = c.Done <- c; // do not block
}
// Terminate pending calls.
client.mutex.Lock();
client.shutdown = err;
for _, call := range client.pending {
call.Error = err;
_ = call.Done <- call; // do not block
}
client.mutex.Unlock();
log.Stderr("client protocol error:", err);
}
// NewClient returns a new Client to handle requests to the
// set of services at the other end of the connection.
func NewClient(conn io.ReadWriteCloser) *Client {
client := new(Client);
client.conn = conn;
client.enc = gob.NewEncoder(conn);
client.dec = gob.NewDecoder(conn);
client.pending = make(map[uint64] *Call);
go client.input();
return client;
}
// DialHTTP connects to an HTTP RPC server at the specified network address.
func DialHTTP(network, address string) (*Client, os.Error) {
conn, err := net.Dial(network, "", address);
if err != nil {
return nil, err
}
io.WriteString(conn, "CONNECT " + rpcPath + " HTTP/1.0\n\n");
// Require successful HTTP response
// before switching to RPC protocol.
resp, err := http.ReadResponse(bufio.NewReader(conn));
if err == nil && resp.Status == connected {
return NewClient(conn), nil;
}
if err == nil {
err = os.ErrorString("unexpected HTTP response: " + resp.Status);
}
conn.Close();
return nil, &net.OpError{"dial-http", network, address, err};
}
// Dial connects to an RPC server at the specified network address.
func Dial(network, address string) (*Client, os.Error) {
conn, err := net.Dial(network, "", address);
if err != nil {
return nil, err
}
return NewClient(conn), nil;
}
// Go invokes the function asynchronously. It returns the Call structure representing
// the invocation. The done channel will signal when the call is complete by returning
// the same Call object. If done is nil, Go will allocate a new channel.
func (client *Client) Go(serviceMethod string, args interface{}, reply interface{}, done chan *Call) *Call {
c := new(Call);
c.ServiceMethod = serviceMethod;
c.Args = args;
c.Reply = reply;
if done == nil {
done = make(chan *Call, 1); // buffered.
} else {
// TODO(r): check cap > 0
// If caller passes done != nil, it must arrange that
// done has enough buffer for the number of simultaneous
// RPCs that will be using that channel.
}
c.Done = done;
if client.shutdown != nil {
c.Error = client.shutdown;
_ = c.Done <- c; // do not block
return c;
}
client.send(c);
return c;
}
// Call invokes the named function, waits for it to complete, and returns its error status.
func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) os.Error {
if client.shutdown != nil {
return client.shutdown
}
call := <-client.Go(serviceMethod, args, reply, nil).Done;
return call.Error;
}
|
