8 "golang.org/x/net/context"
11 // roundTripper manages the request and response from the client-side. A
12 // roundTripper must abide by similar rules to the http.RoundTripper.
13 // Typically, the roundTripper will manage tag assignment and message
15 type roundTripper interface {
16 send(ctx context.Context, msg Message) (Message, error)
19 // transport plays the role of being a client channel manager. It multiplexes
20 // function calls onto the wire and dispatches responses to blocking calls to
21 // send. On the whole, transport is thread-safe for calling send
22 type transport struct {
25 requests chan *fcallRequest
31 var _ roundTripper = &transport{}
33 func newTransport(ctx context.Context, ch *channel) roundTripper {
37 requests: make(chan *fcallRequest),
38 closed: make(chan struct{}),
46 type fcallRequest struct {
53 func newFcallRequest(ctx context.Context, fcall *Fcall) *fcallRequest {
57 response: make(chan *Fcall, 1),
58 err: make(chan error, 1),
62 func (t *transport) send(ctx context.Context, msg Message) (Message, error) {
63 fcall := newFcall(msg)
64 req := newFcallRequest(ctx, fcall)
66 // dispatch the request.
72 case t.requests <- req:
75 // wait for the response.
81 case err := <-req.err:
83 case resp := <-req.response:
84 if resp.Type == Rerror {
85 // pack the error into something useful
86 respmesg, ok := resp.Message.(MessageRerror)
88 return nil, fmt.Errorf("invalid error response: %v", resp)
94 return resp.Message, nil
98 // handle takes messages off the wire and wakes up the waiting tag call.
99 func (t *transport) handle() {
101 log.Println("exited handle loop")
104 // the following variable block are protected components owned by this thread.
106 responses = make(chan *Fcall)
108 // outstanding provides a map of tags to outstanding requests.
109 outstanding = map[Tag]*fcallRequest{}
112 // loop to read messages off of the connection
115 log.Println("exited read loop")
121 if err := t.ch.ReadFcall(t.ctx, fcall); err != nil {
122 switch err := err.(type) {
124 if err.Timeout() || err.Temporary() {
125 // BUG(stevvooe): There may be partial reads under
126 // timeout errors where this is actually fatal.
128 // can only retry if we haven't offset the frame.
133 log.Println("fatal error reading msg:", err)
140 log.Println("ctx done")
143 log.Println("transport closed")
145 case responses <- fcall:
151 log.Println("wait...")
153 case req := <-t.requests:
154 if req.fcall.Tag == NOTAG {
155 // NOTE(stevvooe): We disallow fcalls with NOTAG to come
156 // through this path since we can't join the tagged response
157 // with the waiting caller. This is typically used for the
158 // Tversion/Rversion round trip to setup a session.
160 // It may be better to allow these through but block all
161 // requests until a notag message has a response.
163 req.err <- fmt.Errorf("disallowed tag through transport")
167 // BUG(stevvooe): This is an awful tag allocation procedure.
168 // Replace this with something that let's us allocate tags and
169 // associate data with them, returning to them to a pool when
170 // complete. Such a system would provide a lot of information
171 // about outstanding requests.
174 outstanding[req.fcall.Tag] = req
176 // TODO(stevvooe): Consider the case of requests that never
177 // receive a response. We need to remove the fcall context from
178 // the tag map and dealloc the tag. We may also want to send a
179 // flush for the tag.
180 if err := t.ch.WriteFcall(req.ctx, req.fcall); err != nil {
181 delete(outstanding, req.fcall.Tag)
184 case b := <-responses:
185 req, ok := outstanding[b.Tag]
187 panic("unknown tag received")
189 delete(outstanding, req.fcall.Tag)
193 // TODO(stevvooe): Reclaim tag id.
202 func (t *transport) flush(ctx context.Context, tag Tag) error {
203 // TODO(stevvooe): We need to fire and forget flush messages when a call
204 // context gets cancelled.
205 panic("not implemented")
208 func (t *transport) Close() error {