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
|
// 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";
"rpc";
"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.
type Client struct {
sync.Mutex; // protects pending, seq
closed bool;
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.Lock();
c.seq = client.seq;
client.seq++;
client.pending[c.seq] = c;
client.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) serve() {
var err os.Error;
for err == nil {
response := new(Response);
err = client.dec.Decode(response);
if err != nil {
if err == os.EOF {
break;
}
break;
}
seq := response.Seq;
client.Lock();
c := client.pending[seq];
client.pending[seq] = c, false;
client.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().
doNotBlock := c.Done <- c;
}
client.closed = true;
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.serve();
return client;
}
// Dial 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.
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.closed {
c.Error = os.EOF;
doNotBlock := c.Done <- c;
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.closed {
return os.EOF
}
call := <-client.Go(serviceMethod, args, reply, nil).Done;
return call.Error;
}
|
