13 // TODO(stevvooe): Add net/http.Server-like type here to manage connections.
14 // Coupled with Handler mux, we can get a very http-like experience for 9p
17 // ServeConn the 9p handler over the provided network connection.
18 func ServeConn(ctx context.Context, cn net.Conn, handler Handler) error {
20 // TODO(stevvooe): It would be nice if the handler could declare the
21 // supported version. Before we had handler, we used the session to get
22 // the version (msize, version := session.Version()). We must decided if
23 // we want to proxy version and message size decisions all the back to the
24 // origin server or make those decisions at each link of a proxy chain.
26 ch := newChannel(cn, codec9p{}, DefaultMSize)
27 negctx, cancel := context.WithTimeout(ctx, 1*time.Second)
30 // TODO(stevvooe): For now, we negotiate here. It probably makes sense to
31 // do this outside of this function and then pass in a ready made channel.
32 // We are not really ready to export the channel type yet.
34 if err := servernegotiate(negctx, ch, DefaultVersion); err != nil {
35 // TODO(stevvooe): Need better error handling and retry support here.
36 return fmt.Errorf("error negotiating version: %s", err)
39 ctx = withVersion(ctx, DefaultVersion)
45 closed: make(chan struct{}),
51 // conn plays role of session dispatch for handler in a server.
60 err error // terminal error for the conn
63 // activeRequest includes information about the active request.
64 type activeRequest struct {
67 cancel context.CancelFunc
70 // serve messages on the connection until an error is encountered.
71 func (c *conn) serve() error {
72 tags := map[Tag]*activeRequest{} // active requests
74 requests := make(chan *Fcall) // sync, read-limited
75 responses := make(chan *Fcall) // sync, goroutine consumed
76 completed := make(chan *Fcall) // sync, send in goroutine per request
82 log.Println("server.run()")
85 case req := <-requests:
86 if _, ok := tags[req.Tag]; ok {
88 case responses <- newErrorFcall(req.Tag, ErrDuptag):
89 // Send to responses, bypass tag management.
98 switch msg := req.Message.(type) {
100 log.Println("server: flushing message", msg.Oldtag)
103 // check if we have actually know about the requested flush
104 active, ok := tags[msg.Oldtag]
106 active.cancel() // propagate cancellation to callees
107 delete(tags, msg.Oldtag)
108 resp = newFcall(req.Tag, MessageRflush{})
110 resp = newErrorFcall(req.Tag, ErrUnknownTag)
114 case responses <- resp:
115 // bypass tag management in completed.
122 // Allows us to session handlers to cancel processing of the fcall
124 ctx, cancel := context.WithCancel(c.ctx)
126 // The contents of these instances are only writable in the main
127 // server loop. The value of tag will not change.
128 tags[req.Tag] = &activeRequest{
134 go func(ctx context.Context, req *Fcall) {
135 // TODO(stevvooe): Re-write incoming Treads so that handler
136 // can always respond with a message of the correct msize.
139 msg, err := c.handler.Handle(ctx, req.Message)
141 // all handler errors are forwarded as protocol errors.
142 resp = newErrorFcall(req.Tag, err)
144 resp = newFcall(req.Tag, msg)
148 case completed <- resp:
156 case resp := <-completed:
157 // only responses that flip the tag state traverse this section.
158 active, ok := tags[resp.Tag]
160 // The tag is no longer active. Likely a flushed message.
165 case responses <- resp:
166 case <-active.ctx.Done():
167 // the context was canceled for some reason, perhaps timeout or
168 // due to a flush call. We treat this as a condition where a
169 // response should not be sent.
170 log.Println("canceled", resp, active.ctx.Err())
172 delete(tags, resp.Tag)
181 // read takes requests off the channel and sends them on requests.
182 func (c *conn) read(requests chan *Fcall) {
185 if err := c.ch.ReadFcall(c.ctx, req); err != nil {
186 if err, ok := err.(net.Error); ok {
187 if err.Timeout() || err.Temporary() {
188 // TODO(stevvooe): A full idle timeout on the connection
189 // should be enforced here. No logging because it is quite
195 c.CloseWithError(fmt.Errorf("error reading fcall: %v", err))
200 case requests <- req:
202 c.CloseWithError(c.ctx.Err())
210 func (c *conn) write(responses chan *Fcall) {
213 case resp := <-responses:
214 // TODO(stevvooe): Correctly protect againt overflowing msize from
215 // handler. This can be done above, in the main message handler
216 // loop, by adjusting incoming Tread calls to have a Count that
217 // won't overflow the msize.
219 if err := c.ch.WriteFcall(c.ctx, resp); err != nil {
220 if err, ok := err.(net.Error); ok {
221 if err.Timeout() || err.Temporary() {
222 // TODO(stevvooe): A full idle timeout on the
223 // connection should be enforced here. We log here,
224 // since this is less common.
225 log.Printf("9p server: temporary error writing fcall: %v", err)
230 c.CloseWithError(fmt.Errorf("error writing fcall: %v", err))
234 c.CloseWithError(c.ctx.Err())
242 func (c *conn) Close() error {
243 return c.CloseWithError(nil)
246 func (c *conn) CloseWithError(err error) error {