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