libgo/go/http/transport.go

   1 // Copyright 2011 The Go Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style
   3 // license that can be found in the LICENSE file.
   4
   5 package http
   6
   7 import (
   8         "bufio"
   9         "compress/gzip"
  10         "crypto/tls"
  11         "encoding/base64"
  12         "fmt"
  13         "io"
  14         "io/ioutil"
  15         "log"
  16         "net"
  17         "os"
  18         "strings"
  19         "sync"
  20         "url"
  21 )
  22
  23 // DefaultTransport is the default implementation of Transport and is
  24 // used by DefaultClient.  It establishes a new network connection for
  25 // each call to Do and uses HTTP proxies as directed by the
  26 // $HTTP_PROXY and $NO_PROXY (or $http_proxy and $no_proxy)
  27 // environment variables.
  28 var DefaultTransport RoundTripper = &Transport{Proxy: ProxyFromEnvironment}
  29
  30 // DefaultMaxIdleConnsPerHost is the default value of Transport's
  31 // MaxIdleConnsPerHost.
  32 const DefaultMaxIdleConnsPerHost = 2
  33
  34 // Transport is an implementation of RoundTripper that supports http,
  35 // https, and http proxies (for either http or https with CONNECT).
  36 // Transport can also cache connections for future re-use.
  37 type Transport struct {
  38         lk       sync.Mutex
  39         idleConn map[string][]*persistConn
  40         altProto map[string]RoundTripper // nil or map of URI scheme => RoundTripper
  41
  42         // TODO: tunable on global max cached connections
  43         // TODO: tunable on timeout on cached connections
  44         // TODO: optional pipelining
  45
  46         // Proxy specifies a function to return a proxy for a given
  47         // Request. If the function returns a non-nil error, the
  48         // request is aborted with the provided error.
  49         // If Proxy is nil or returns a nil *URL, no proxy is used.
  50         Proxy func(*Request) (*url.URL, os.Error)
  51
  52         // Dial specifies the dial function for creating TCP
  53         // connections.
  54         // If Dial is nil, net.Dial is used.
  55         Dial func(net, addr string) (c net.Conn, err os.Error)
  56
  57         // TLSClientConfig specifies the TLS configuration to use with
  58         // tls.Client. If nil, the default configuration is used.
  59         TLSClientConfig *tls.Config
  60
  61         DisableKeepAlives  bool
  62         DisableCompression bool
  63
  64         // MaxIdleConnsPerHost, if non-zero, controls the maximum idle
  65         // (keep-alive) to keep to keep per-host.  If zero,
  66         // DefaultMaxIdleConnsPerHost is used.
  67         MaxIdleConnsPerHost int
  68 }
  69
  70 // ProxyFromEnvironment returns the URL of the proxy to use for a
  71 // given request, as indicated by the environment variables
  72 // $HTTP_PROXY and $NO_PROXY (or $http_proxy and $no_proxy).
  73 // Either URL or an error is returned.
  74 func ProxyFromEnvironment(req *Request) (*url.URL, os.Error) {
  75         proxy := getenvEitherCase("HTTP_PROXY")
  76         if proxy == "" {
  77                 return nil, nil
  78         }
  79         if !useProxy(canonicalAddr(req.URL)) {
  80                 return nil, nil
  81         }
  82         proxyURL, err := url.ParseRequest(proxy)
  83         if err != nil {
  84                 return nil, os.NewError("invalid proxy address")
  85         }
  86         if proxyURL.Host == "" {
  87                 proxyURL, err = url.ParseRequest("http://" + proxy)
  88                 if err != nil {
  89                         return nil, os.NewError("invalid proxy address")
  90                 }
  91         }
  92         return proxyURL, nil
  93 }
  94
  95 // ProxyURL returns a proxy function (for use in a Transport)
  96 // that always returns the same URL.
  97 func ProxyURL(fixedURL *url.URL) func(*Request) (*url.URL, os.Error) {
  98         return func(*Request) (*url.URL, os.Error) {
  99                 return fixedURL, nil
 100         }
 101 }
 102
 103 // transportRequest is a wrapper around a *Request that adds
 104 // optional extra headers to write.
 105 type transportRequest struct {
 106         *Request        // original request, not to be mutated
 107         extra    Header // extra headers to write, or nil
 108 }
 109
 110 func (tr *transportRequest) extraHeaders() Header {
 111         if tr.extra == nil {
 112                 tr.extra = make(Header)
 113         }
 114         return tr.extra
 115 }
 116
 117 // RoundTrip implements the RoundTripper interface.
 118 func (t *Transport) RoundTrip(req *Request) (resp *Response, err os.Error) {
 119         if req.URL == nil {
 120                 return nil, os.NewError("http: nil Request.URL")
 121         }
 122         if req.Header == nil {
 123                 return nil, os.NewError("http: nil Request.Header")
 124         }
 125         if req.URL.Scheme != "http" && req.URL.Scheme != "https" {
 126                 t.lk.Lock()
 127                 var rt RoundTripper
 128                 if t.altProto != nil {
 129                         rt = t.altProto[req.URL.Scheme]
 130                 }
 131                 t.lk.Unlock()
 132                 if rt == nil {
 133                         return nil, &badStringError{"unsupported protocol scheme", req.URL.Scheme}
 134                 }
 135                 return rt.RoundTrip(req)
 136         }
 137         treq := &transportRequest{Request: req}
 138         cm, err := t.connectMethodForRequest(treq)
 139         if err != nil {
 140                 return nil, err
 141         }
 142
 143         // Get the cached or newly-created connection to either the
 144         // host (for http or https), the http proxy, or the http proxy
 145         // pre-CONNECTed to https server.  In any case, we'll be ready
 146         // to send it requests.
 147         pconn, err := t.getConn(cm)
 148         if err != nil {
 149                 return nil, err
 150         }
 151
 152         return pconn.roundTrip(treq)
 153 }
 154
 155 // RegisterProtocol registers a new protocol with scheme.
 156 // The Transport will pass requests using the given scheme to rt.
 157 // It is rt's responsibility to simulate HTTP request semantics.
 158 //
 159 // RegisterProtocol can be used by other packages to provide
 160 // implementations of protocol schemes like "ftp" or "file".
 161 func (t *Transport) RegisterProtocol(scheme string, rt RoundTripper) {
 162         if scheme == "http" || scheme == "https" {
 163                 panic("protocol " + scheme + " already registered")
 164         }
 165         t.lk.Lock()
 166         defer t.lk.Unlock()
 167         if t.altProto == nil {
 168                 t.altProto = make(map[string]RoundTripper)
 169         }
 170         if _, exists := t.altProto[scheme]; exists {
 171                 panic("protocol " + scheme + " already registered")
 172         }
 173         t.altProto[scheme] = rt
 174 }
 175
 176 // CloseIdleConnections closes any connections which were previously
 177 // connected from previous requests but are now sitting idle in
 178 // a "keep-alive" state. It does not interrupt any connections currently
 179 // in use.
 180 func (t *Transport) CloseIdleConnections() {
 181         t.lk.Lock()
 182         defer t.lk.Unlock()
 183         if t.idleConn == nil {
 184                 return
 185         }
 186         for _, conns := range t.idleConn {
 187                 for _, pconn := range conns {
 188                         pconn.close()
 189                 }
 190         }
 191         t.idleConn = nil
 192 }
 193
 194 //
 195 // Private implementation past this point.
 196 //
 197
 198 func getenvEitherCase(k string) string {
 199         if v := os.Getenv(strings.ToUpper(k)); v != "" {
 200                 return v
 201         }
 202         return os.Getenv(strings.ToLower(k))
 203 }
 204
 205 func (t *Transport) connectMethodForRequest(treq *transportRequest) (*connectMethod, os.Error) {
 206         cm := &connectMethod{
 207                 targetScheme: treq.URL.Scheme,
 208                 targetAddr:   canonicalAddr(treq.URL),
 209         }
 210         if t.Proxy != nil {
 211                 var err os.Error
 212                 cm.proxyURL, err = t.Proxy(treq.Request)
 213                 if err != nil {
 214                         return nil, err
 215                 }
 216         }
 217         return cm, nil
 218 }
 219
 220 // proxyAuth returns the Proxy-Authorization header to set
 221 // on requests, if applicable.
 222 func (cm *connectMethod) proxyAuth() string {
 223         if cm.proxyURL == nil {
 224                 return ""
 225         }
 226         proxyInfo := cm.proxyURL.RawUserinfo
 227         if proxyInfo != "" {
 228                 return "Basic " + base64.URLEncoding.EncodeToString([]byte(proxyInfo))
 229         }
 230         return ""
 231 }
 232
 233 func (t *Transport) putIdleConn(pconn *persistConn) {
 234         t.lk.Lock()
 235         defer t.lk.Unlock()
 236         if t.DisableKeepAlives || t.MaxIdleConnsPerHost < 0 {
 237                 pconn.close()
 238                 return
 239         }
 240         if pconn.isBroken() {
 241                 return
 242         }
 243         key := pconn.cacheKey
 244         max := t.MaxIdleConnsPerHost
 245         if max == 0 {
 246                 max = DefaultMaxIdleConnsPerHost
 247         }
 248         if len(t.idleConn[key]) >= max {
 249                 pconn.close()
 250                 return
 251         }
 252         t.idleConn[key] = append(t.idleConn[key], pconn)
 253 }
 254
 255 func (t *Transport) getIdleConn(cm *connectMethod) (pconn *persistConn) {
 256         t.lk.Lock()
 257         defer t.lk.Unlock()
 258         if t.idleConn == nil {
 259                 t.idleConn = make(map[string][]*persistConn)
 260         }
 261         key := cm.String()
 262         for {
 263                 pconns, ok := t.idleConn[key]
 264                 if !ok {
 265                         return nil
 266                 }
 267                 if len(pconns) == 1 {
 268                         pconn = pconns[0]
 269                         delete(t.idleConn, key)
 270                 } else {
 271                         // 2 or more cached connections; pop last
 272                         // TODO: queue?
 273                         pconn = pconns[len(pconns)-1]
 274                         t.idleConn[key] = pconns[0 : len(pconns)-1]
 275                 }
 276                 if !pconn.isBroken() {
 277                         return
 278                 }
 279         }
 280         return
 281 }
 282
 283 func (t *Transport) dial(network, addr string) (c net.Conn, err os.Error) {
 284         if t.Dial != nil {
 285                 return t.Dial(network, addr)
 286         }
 287         return net.Dial(network, addr)
 288 }
 289
 290 // getConn dials and creates a new persistConn to the target as
 291 // specified in the connectMethod.  This includes doing a proxy CONNECT
 292 // and/or setting up TLS.  If this doesn't return an error, the persistConn
 293 // is ready to write requests to.
 294 func (t *Transport) getConn(cm *connectMethod) (*persistConn, os.Error) {
 295         if pc := t.getIdleConn(cm); pc != nil {
 296                 return pc, nil
 297         }
 298
 299         conn, err := t.dial("tcp", cm.addr())
 300         if err != nil {
 301                 if cm.proxyURL != nil {
 302                         err = fmt.Errorf("http: error connecting to proxy %s: %v", cm.proxyURL, err)
 303                 }
 304                 return nil, err
 305         }
 306
 307         pa := cm.proxyAuth()
 308
 309         pconn := &persistConn{
 310                 t:        t,
 311                 cacheKey: cm.String(),
 312                 conn:     conn,
 313                 reqch:    make(chan requestAndChan, 50),
 314         }
 315
 316         switch {
 317         case cm.proxyURL == nil:
 318                 // Do nothing.
 319         case cm.targetScheme == "http":
 320                 pconn.isProxy = true
 321                 if pa != "" {
 322                         pconn.mutateHeaderFunc = func(h Header) {
 323                                 h.Set("Proxy-Authorization", pa)
 324                         }
 325                 }
 326         case cm.targetScheme == "https":
 327                 connectReq := &Request{
 328                         Method: "CONNECT",
 329                         URL:    &url.URL{RawPath: cm.targetAddr},
 330                         Host:   cm.targetAddr,
 331                         Header: make(Header),
 332                 }
 333                 if pa != "" {
 334                         connectReq.Header.Set("Proxy-Authorization", pa)
 335                 }
 336                 connectReq.Write(conn)
 337
 338                 // Read response.
 339                 // Okay to use and discard buffered reader here, because
 340                 // TLS server will not speak until spoken to.
 341                 br := bufio.NewReader(conn)
 342                 resp, err := ReadResponse(br, connectReq)
 343                 if err != nil {
 344                         conn.Close()
 345                         return nil, err
 346                 }
 347                 if resp.StatusCode != 200 {
 348                         f := strings.SplitN(resp.Status, " ", 2)
 349                         conn.Close()
 350                         return nil, os.NewError(f[1])
 351                 }
 352         }
 353
 354         if cm.targetScheme == "https" {
 355                 // Initiate TLS and check remote host name against certificate.
 356                 conn = tls.Client(conn, t.TLSClientConfig)
 357                 if err = conn.(*tls.Conn).Handshake(); err != nil {
 358                         return nil, err
 359                 }
 360                 if err = conn.(*tls.Conn).VerifyHostname(cm.tlsHost()); err != nil {
 361                         return nil, err
 362                 }
 363                 pconn.conn = conn
 364         }
 365
 366         pconn.br = bufio.NewReader(pconn.conn)
 367         pconn.cc = NewClientConn(conn, pconn.br)
 368         go pconn.readLoop()
 369         return pconn, nil
 370 }
 371
 372 // useProxy returns true if requests to addr should use a proxy,
 373 // according to the NO_PROXY or no_proxy environment variable.
 374 // addr is always a canonicalAddr with a host and port.
 375 func useProxy(addr string) bool {
 376         if len(addr) == 0 {
 377                 return true
 378         }
 379         host, _, err := net.SplitHostPort(addr)
 380         if err != nil {
 381                 return false
 382         }
 383         if host == "localhost" {
 384                 return false
 385         }
 386         if ip := net.ParseIP(host); ip != nil {
 387                 if ip.IsLoopback() {
 388                         return false
 389                 }
 390         }
 391
 392         no_proxy := getenvEitherCase("NO_PROXY")
 393         if no_proxy == "*" {
 394                 return false
 395         }
 396
 397         addr = strings.ToLower(strings.TrimSpace(addr))
 398         if hasPort(addr) {
 399                 addr = addr[:strings.LastIndex(addr, ":")]
 400         }
 401
 402         for _, p := range strings.Split(no_proxy, ",") {
 403                 p = strings.ToLower(strings.TrimSpace(p))
 404                 if len(p) == 0 {
 405                         continue
 406                 }
 407                 if hasPort(p) {
 408                         p = p[:strings.LastIndex(p, ":")]
 409                 }
 410                 if addr == p || (p[0] == '.' && (strings.HasSuffix(addr, p) || addr == p[1:])) {
 411                         return false
 412                 }
 413         }
 414         return true
 415 }
 416
 417 // connectMethod is the map key (in its String form) for keeping persistent
 418 // TCP connections alive for subsequent HTTP requests.
 419 //
 420 // A connect method may be of the following types:
 421 //
 422 // Cache key form                Description
 423 // -----------------             -------------------------
 424 // ||http|foo.com                http directly to server, no proxy
 425 // ||https|foo.com               https directly to server, no proxy
 426 // http://proxy.com|https|foo.com  http to proxy, then CONNECT to foo.com
 427 // http://proxy.com|http           http to proxy, http to anywhere after that
 428 //
 429 // Note: no support to https to the proxy yet.
 430 //
 431 type connectMethod struct {
 432         proxyURL     *url.URL // nil for no proxy, else full proxy URL
 433         targetScheme string   // "http" or "https"
 434         targetAddr   string   // Not used if proxy + http targetScheme (4th example in table)
 435 }
 436
 437 func (ck *connectMethod) String() string {
 438         proxyStr := ""
 439         if ck.proxyURL != nil {
 440                 proxyStr = ck.proxyURL.String()
 441         }
 442         return strings.Join([]string{proxyStr, ck.targetScheme, ck.targetAddr}, "|")
 443 }
 444
 445 // addr returns the first hop "host:port" to which we need to TCP connect.
 446 func (cm *connectMethod) addr() string {
 447         if cm.proxyURL != nil {
 448                 return canonicalAddr(cm.proxyURL)
 449         }
 450         return cm.targetAddr
 451 }
 452
 453 // tlsHost returns the host name to match against the peer's
 454 // TLS certificate.
 455 func (cm *connectMethod) tlsHost() string {
 456         h := cm.targetAddr
 457         if hasPort(h) {
 458                 h = h[:strings.LastIndex(h, ":")]
 459         }
 460         return h
 461 }
 462
 463 // persistConn wraps a connection, usually a persistent one
 464 // (but may be used for non-keep-alive requests as well)
 465 type persistConn struct {
 466         t        *Transport
 467         cacheKey string // its connectMethod.String()
 468         conn     net.Conn
 469         cc       *ClientConn
 470         br       *bufio.Reader
 471         reqch    chan requestAndChan // written by roundTrip(); read by readLoop()
 472         isProxy  bool
 473
 474         // mutateHeaderFunc is an optional func to modify extra
 475         // headers on each outbound request before it's written. (the
 476         // original Request given to RoundTrip is not modified)
 477         mutateHeaderFunc func(Header)
 478
 479         lk                   sync.Mutex // guards numExpectedResponses and broken
 480         numExpectedResponses int
 481         broken               bool // an error has happened on this connection; marked broken so it's not reused.
 482 }
 483
 484 func (pc *persistConn) isBroken() bool {
 485         pc.lk.Lock()
 486         defer pc.lk.Unlock()
 487         return pc.broken
 488 }
 489
 490 func (pc *persistConn) expectingResponse() bool {
 491         pc.lk.Lock()
 492         defer pc.lk.Unlock()
 493         return pc.numExpectedResponses > 0
 494 }
 495
 496 var remoteSideClosedFunc func(os.Error) bool // or nil to use default
 497
 498 func remoteSideClosed(err os.Error) bool {
 499         if err == os.EOF || err == os.EINVAL {
 500                 return true
 501         }
 502         if remoteSideClosedFunc != nil {
 503                 return remoteSideClosedFunc(err)
 504         }
 505         return false
 506 }
 507
 508 func (pc *persistConn) readLoop() {
 509         alive := true
 510         for alive {
 511                 pb, err := pc.br.Peek(1)
 512                 if err != nil {
 513                         if remoteSideClosed(err) && !pc.expectingResponse() {
 514                                 // Remote side closed on us.  (We probably hit their
 515                                 // max idle timeout)
 516                                 pc.close()
 517                                 return
 518                         }
 519                 }
 520                 if !pc.expectingResponse() {
 521                         log.Printf("Unsolicited response received on idle HTTP channel starting with %q; err=%v",
 522                                 string(pb), err)
 523                         pc.close()
 524                         return
 525                 }
 526
 527                 rc := <-pc.reqch
 528                 resp, err := pc.cc.readUsing(rc.req, func(buf *bufio.Reader, forReq *Request) (*Response, os.Error) {
 529                         resp, err := ReadResponse(buf, forReq)
 530                         if err != nil || resp.ContentLength == 0 {
 531                                 return resp, err
 532                         }
 533                         if rc.addedGzip && resp.Header.Get("Content-Encoding") == "gzip" {
 534                                 resp.Header.Del("Content-Encoding")
 535                                 resp.Header.Del("Content-Length")
 536                                 resp.ContentLength = -1
 537                                 gzReader, err := gzip.NewReader(resp.Body)
 538                                 if err != nil {
 539                                         pc.close()
 540                                         return nil, err
 541                                 }
 542                                 resp.Body = &readFirstCloseBoth{&discardOnCloseReadCloser{gzReader}, resp.Body}
 543                         }
 544                         resp.Body = &bodyEOFSignal{body: resp.Body}
 545                         return resp, err
 546                 })
 547
 548                 if err == ErrPersistEOF {
 549                         // Succeeded, but we can't send any more
 550                         // persistent connections on this again.  We
 551                         // hide this error to upstream callers.
 552                         alive = false
 553                         err = nil
 554                 } else if err != nil || rc.req.Close {
 555                         alive = false
 556                 }
 557
 558                 hasBody := resp != nil && resp.ContentLength != 0
 559                 var waitForBodyRead chan bool
 560                 if alive {
 561                         if hasBody {
 562                                 waitForBodyRead = make(chan bool)
 563                                 resp.Body.(*bodyEOFSignal).fn = func() {
 564                                         pc.t.putIdleConn(pc)
 565                                         waitForBodyRead <- true
 566                                 }
 567                         } else {
 568                                 // When there's no response body, we immediately
 569                                 // reuse the TCP connection (putIdleConn), but
 570                                 // we need to prevent ClientConn.Read from
 571                                 // closing the Response.Body on the next
 572                                 // loop, otherwise it might close the body
 573                                 // before the client code has had a chance to
 574                                 // read it (even though it'll just be 0, EOF).
 575                                 pc.cc.lk.Lock()
 576                                 pc.cc.lastbody = nil
 577                                 pc.cc.lk.Unlock()
 578
 579                                 pc.t.putIdleConn(pc)
 580                         }
 581                 }
 582
 583                 rc.ch <- responseAndError{resp, err}
 584
 585                 // Wait for the just-returned response body to be fully consumed
 586                 // before we race and peek on the underlying bufio reader.
 587                 if waitForBodyRead != nil {
 588                         <-waitForBodyRead
 589                 }
 590         }
 591 }
 592
 593 type responseAndError struct {
 594         res *Response
 595         err os.Error
 596 }
 597
 598 type requestAndChan struct {
 599         req *Request
 600         ch  chan responseAndError
 601
 602         // did the Transport (as opposed to the client code) add an
 603         // Accept-Encoding gzip header? only if it we set it do
 604         // we transparently decode the gzip.
 605         addedGzip bool
 606 }
 607
 608 func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err os.Error) {
 609         if pc.mutateHeaderFunc != nil {
 610                 pc.mutateHeaderFunc(req.extraHeaders())
 611         }
 612
 613         // Ask for a compressed version if the caller didn't set their
 614         // own value for Accept-Encoding. We only attempted to
 615         // uncompress the gzip stream if we were the layer that
 616         // requested it.
 617         requestedGzip := false
 618         if !pc.t.DisableCompression && req.Header.Get("Accept-Encoding") == "" {
 619                 // Request gzip only, not deflate. Deflate is ambiguous and
 620                 // not as universally supported anyway.
 621                 // See: http://www.gzip.org/zlib/zlib_faq.html#faq38
 622                 requestedGzip = true
 623                 req.extraHeaders().Set("Accept-Encoding", "gzip")
 624         }
 625
 626         pc.lk.Lock()
 627         pc.numExpectedResponses++
 628         pc.lk.Unlock()
 629
 630         pc.cc.writeReq = func(r *Request, w io.Writer) os.Error {
 631                 return r.write(w, pc.isProxy, req.extra)
 632         }
 633
 634         err = pc.cc.Write(req.Request)
 635         if err != nil {
 636                 pc.close()
 637                 return
 638         }
 639
 640         ch := make(chan responseAndError, 1)
 641         pc.reqch <- requestAndChan{req.Request, ch, requestedGzip}
 642         re := <-ch
 643         pc.lk.Lock()
 644         pc.numExpectedResponses--
 645         pc.lk.Unlock()
 646
 647         return re.res, re.err
 648 }
 649
 650 func (pc *persistConn) close() {
 651         pc.lk.Lock()
 652         defer pc.lk.Unlock()
 653         pc.broken = true
 654         pc.cc.Close()
 655         pc.conn.Close()
 656         pc.mutateHeaderFunc = nil
 657 }
 658
 659 var portMap = map[string]string{
 660         "http":  "80",
 661         "https": "443",
 662 }
 663
 664 // canonicalAddr returns url.Host but always with a ":port" suffix
 665 func canonicalAddr(url *url.URL) string {
 666         addr := url.Host
 667         if !hasPort(addr) {
 668                 return addr + ":" + portMap[url.Scheme]
 669         }
 670         return addr
 671 }
 672
 673 func responseIsKeepAlive(res *Response) bool {
 674         // TODO: implement.  for now just always shutting down the connection.
 675         return false
 676 }
 677
 678 // bodyEOFSignal wraps a ReadCloser but runs fn (if non-nil) at most
 679 // once, right before the final Read() or Close() call returns, but after
 680 // EOF has been seen.
 681 type bodyEOFSignal struct {
 682         body     io.ReadCloser
 683         fn       func()
 684         isClosed bool
 685 }
 686
 687 func (es *bodyEOFSignal) Read(p []byte) (n int, err os.Error) {
 688         n, err = es.body.Read(p)
 689         if es.isClosed && n > 0 {
 690                 panic("http: unexpected bodyEOFSignal Read after Close; see issue 1725")
 691         }
 692         if err == os.EOF && es.fn != nil {
 693                 es.fn()
 694                 es.fn = nil
 695         }
 696         return
 697 }
 698
 699 func (es *bodyEOFSignal) Close() (err os.Error) {
 700         if es.isClosed {
 701                 return nil
 702         }
 703         es.isClosed = true
 704         err = es.body.Close()
 705         if err == nil && es.fn != nil {
 706                 es.fn()
 707                 es.fn = nil
 708         }
 709         return
 710 }
 711
 712 type readFirstCloseBoth struct {
 713         io.ReadCloser
 714         io.Closer
 715 }
 716
 717 func (r *readFirstCloseBoth) Close() os.Error {
 718         if err := r.ReadCloser.Close(); err != nil {
 719                 r.Closer.Close()
 720                 return err
 721         }
 722         if err := r.Closer.Close(); err != nil {
 723                 return err
 724         }
 725         return nil
 726 }
 727
 728 // discardOnCloseReadCloser consumes all its input on Close.
 729 type discardOnCloseReadCloser struct {
 730         io.ReadCloser
 731 }
 732
 733 func (d *discardOnCloseReadCloser) Close() os.Error {
 734         io.Copy(ioutil.Discard, d.ReadCloser) // ignore errors; likely invalid or already closed
 735         return d.ReadCloser.Close()
 736 }