http: persistent connection objects

R=rsc
CC=golang-dev
https://golang.org/cl/203051

src/pkg/http/Makefile

@@ -10,6 +10,7 @@ GOFILES=\
 	client.go\
 	fs.go\
 	lex.go\
+	persist.go\
 	request.go\
 	response.go\
 	server.go\

src/pkg/http/persist.go

@@ -0,0 +1,285 @@
+// 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 http
+
+import (
+	"bufio"
+	"container/list"
+	"io"
+	"net"
+	"os"
+	"sync"
+)
+
+var ErrPersistEOF = &ProtocolError{"persistent connection closed"}
+
+// A ServerConn reads requests and sends responses over an underlying
+// connection, until the HTTP keepalive logic commands an end. ServerConn
+// does not close the underlying connection. Instead, the user calls Close
+// and regains control over the connection. ServerConn supports pipe-lining,
+// i.e. requests can be read out of sync (but in the same order) while the
+// respective responses are sent.
+type ServerConn struct {
+	c               net.Conn
+	r               *bufio.Reader
+	clsd            bool     // indicates a graceful close
+	re, we          os.Error // read/write errors
+	lastBody        io.ReadCloser
+	nread, nwritten int
+	lk              sync.Mutex // protected read/write to re,we
+}
+
+// NewServerConn returns a new ServerConn reading and writing c.  If r is not
+// nil, it is the buffer to use when reading c.
+func NewServerConn(c net.Conn, r *bufio.Reader) *ServerConn {
+	if r == nil {
+		r = bufio.NewReader(c)
+	}
+	return &ServerConn{c: c, r: r}
+}
+
+// Close detaches the ServerConn and returns the underlying connection as well
+// as the read-side bufio which may have some left over data. Close may be
+// called before Read has signaled the end of the keep-alive logic. The user
+// should not call Close while Read or Write is in progress.
+func (sc *ServerConn) Close() (c net.Conn, r *bufio.Reader) {
+	c = sc.c
+	r = sc.r
+	sc.c = nil
+	sc.r = nil
+	return
+}
+
+// Read returns the next request on the wire. An ErrPersistEOF is returned if
+// it is gracefully determined that there are no more requests (e.g. after the
+// first request on an HTTP/1.0 connection, or after a Connection:close on a
+// HTTP/1.1 connection). Read can be called concurrently with Write, but not
+// with another Read.
+func (sc *ServerConn) Read() (req *Request, err os.Error) {
+
+	sc.lk.Lock()
+	if sc.we != nil { // no point receiving if write-side broken or closed
+		defer sc.lk.Unlock()
+		return nil, sc.we
+	}
+	if sc.re != nil {
+		defer sc.lk.Unlock()
+		return nil, sc.re
+	}
+	sc.lk.Unlock()
+
+	// Make sure body is fully consumed, even if user does not call body.Close
+	if sc.lastBody != nil {
+		// body.Close is assumed to be idempotent and multiple calls to
+		// it should return the error that its first invokation
+		// returned.
+		err = sc.lastBody.Close()
+		sc.lastBody = nil
+		if err != nil {
+			sc.lk.Lock()
+			defer sc.lk.Unlock()
+			sc.re = err
+			return nil, err
+		}
+	}
+
+	req, err = ReadRequest(sc.r)
+	if err != nil {
+		sc.lk.Lock()
+		defer sc.lk.Unlock()
+		if err == io.ErrUnexpectedEOF {
+			// A close from the opposing client is treated as a
+			// graceful close, even if there was some unparse-able
+			// data before the close.
+			sc.re = ErrPersistEOF
+			return nil, sc.re
+		} else {
+			sc.re = err
+			return
+		}
+	}
+	sc.lastBody = req.Body
+	sc.nread++
+	if req.Close {
+		sc.lk.Lock()
+		defer sc.lk.Unlock()
+		sc.re = ErrPersistEOF
+		return req, sc.re
+	}
+	return
+}
+
+// Write writes a repsonse. To close the connection gracefully, set the
+// Response.Close field to true. Write should be considered operational until
+// it returns an error, regardless of any errors returned on the Read side.
+// Write can be called concurrently with Read, but not with another Write.
+func (sc *ServerConn) Write(resp *Response) os.Error {
+
+	sc.lk.Lock()
+	if sc.we != nil {
+		defer sc.lk.Unlock()
+		return sc.we
+	}
+	sc.lk.Unlock()
+	if sc.nread <= sc.nwritten {
+		return os.NewError("persist server pipe count")
+	}
+
+	if resp.Close {
+		// After signaling a keep-alive close, any pipelined unread
+		// requests will be lost. It is up to the user to drain them
+		// before signaling.
+		sc.lk.Lock()
+		sc.re = ErrPersistEOF
+		sc.lk.Unlock()
+	}
+
+	err := resp.Write(sc.c)
+	if err != nil {
+		sc.lk.Lock()
+		defer sc.lk.Unlock()
+		sc.we = err
+		return err
+	}
+	sc.nwritten++
+
+	return nil
+}
+
+// A ClientConn sends request and receives headers over an underlying
+// connection, while respecting the HTTP keepalive logic. ClientConn is not
+// responsible for closing the underlying connection. One must call Close to
+// regain control of that connection and deal with it as desired.
+type ClientConn struct {
+	c               net.Conn
+	r               *bufio.Reader
+	re, we          os.Error // read/write errors
+	lastBody        io.ReadCloser
+	nread, nwritten int
+	reqm            list.List  // request methods in order of execution
+	lk              sync.Mutex // protects read/write to reqm,re,we
+}
+
+// NewClientConn returns a new ClientConn reading and writing c.  If r is not
+// nil, it is the buffer to use when reading c.
+func NewClientConn(c net.Conn, r *bufio.Reader) *ClientConn {
+	if r == nil {
+		r = bufio.NewReader(c)
+	}
+	return &ClientConn{c: c, r: r}
+}
+
+// Close detaches the ClientConn and returns the underlying connection as well
+// as the read-side bufio which may have some left over data. Close may be
+// called before the user or Read have signaled the end of the keep-alive
+// logic. The user should not call Close while Read or Write is in progress.
+func (cc *ClientConn) Close() (c net.Conn, r *bufio.Reader) {
+	c = cc.c
+	r = cc.r
+	cc.c = nil
+	cc.r = nil
+	cc.lk.Lock()
+	cc.reqm.Init()
+	cc.lk.Unlock()
+	return
+}
+
+// Write writes a request. An ErrPersistEOF error is returned if the connection
+// has been closed in an HTTP keepalive sense. If req.Close equals true, the
+// keepalive connection is logically closed after this request and the opposing
+// server is informed. An ErrUnexpectedEOF indicates the remote closed the
+// underlying TCP connection, which is usually considered as graceful close.
+// Write can be called concurrently with Read, but not with another Write.
+func (cc *ClientConn) Write(req *Request) os.Error {
+
+	cc.lk.Lock()
+	if cc.re != nil { // no point sending if read-side closed or broken
+		defer cc.lk.Unlock()
+		return cc.re
+	}
+	if cc.we != nil {
+		defer cc.lk.Unlock()
+		return cc.we
+	}
+	cc.lk.Unlock()
+
+	if req.Close {
+		// We write the EOF to the write-side error, because there
+		// still might be some pipelined reads
+		cc.lk.Lock()
+		cc.we = ErrPersistEOF
+		cc.lk.Unlock()
+	}
+
+	err := req.Write(cc.c)
+	if err != nil {
+		cc.lk.Lock()
+		defer cc.lk.Unlock()
+		cc.we = err
+		return err
+	}
+	cc.nwritten++
+	cc.lk.Lock()
+	cc.reqm.PushBack(req.Method)
+	cc.lk.Unlock()
+
+	return nil
+}
+
+// Read reads the next response from the wire. A valid response might be
+// returned together with an ErrPersistEOF, which means that the remote
+// requested that this be the last request serviced. Read can be called
+// concurrently with Write, but not with another Read.
+func (cc *ClientConn) Read() (resp *Response, err os.Error) {
+
+	cc.lk.Lock()
+	if cc.re != nil {
+		defer cc.lk.Unlock()
+		return nil, cc.re
+	}
+	cc.lk.Unlock()
+
+	if cc.nread >= cc.nwritten {
+		return nil, os.NewError("persist client pipe count")
+	}
+
+	// Make sure body is fully consumed, even if user does not call body.Close
+	if cc.lastBody != nil {
+		// body.Close is assumed to be idempotent and multiple calls to
+		// it should return the error that its first invokation
+		// returned.
+		err = cc.lastBody.Close()
+		cc.lastBody = nil
+		if err != nil {
+			cc.lk.Lock()
+			defer cc.lk.Unlock()
+			cc.re = err
+			return nil, err
+		}
+	}
+
+	cc.lk.Lock()
+	m := cc.reqm.Front()
+	cc.reqm.Remove(m)
+	cc.lk.Unlock()
+	resp, err = ReadResponse(cc.r, m.Value.(string))
+	if err != nil {
+		cc.lk.Lock()
+		defer cc.lk.Unlock()
+		cc.re = err
+		return
+	}
+	cc.lastBody = resp.Body
+
+	cc.nread++
+
+	if resp.Close {
+		cc.lk.Lock()
+		defer cc.lk.Unlock()
+		cc.re = ErrPersistEOF // don't send any more requests
+		return resp, cc.re
+	}
+	return
+}