Merge pull request #7269 from sinsharat/use_requestWithContext_for_cancel
*: Use http.Request.WithContext instead of Cancelrelease-3.2
commit
56c706ff91
|
@ -16,6 +16,7 @@ package leasehttp
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
|
@ -26,7 +27,6 @@ import (
|
||||||
"github.com/coreos/etcd/lease"
|
"github.com/coreos/etcd/lease"
|
||||||
"github.com/coreos/etcd/lease/leasepb"
|
"github.com/coreos/etcd/lease/leasepb"
|
||||||
"github.com/coreos/etcd/pkg/httputil"
|
"github.com/coreos/etcd/pkg/httputil"
|
||||||
"golang.org/x/net/context"
|
|
||||||
)
|
)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
|
@ -202,45 +202,27 @@ func TimeToLiveHTTP(ctx context.Context, id lease.LeaseID, keys bool, url string
|
||||||
}
|
}
|
||||||
req.Header.Set("Content-Type", "application/protobuf")
|
req.Header.Set("Content-Type", "application/protobuf")
|
||||||
|
|
||||||
cancel := httputil.RequestCanceler(req)
|
req = req.WithContext(ctx)
|
||||||
|
|
||||||
cc := &http.Client{Transport: rt}
|
cc := &http.Client{Transport: rt}
|
||||||
var b []byte
|
var b []byte
|
||||||
// buffer errc channel so that errc don't block inside the go routinue
|
// buffer errc channel so that errc don't block inside the go routinue
|
||||||
errc := make(chan error, 2)
|
|
||||||
go func() {
|
|
||||||
resp, err := cc.Do(req)
|
resp, err := cc.Do(req)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errc <- err
|
return nil, err
|
||||||
return
|
|
||||||
}
|
}
|
||||||
b, err = readResponse(resp)
|
b, err = readResponse(resp)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
errc <- err
|
return nil, err
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if resp.StatusCode == http.StatusRequestTimeout {
|
if resp.StatusCode == http.StatusRequestTimeout {
|
||||||
errc <- ErrLeaseHTTPTimeout
|
return nil, ErrLeaseHTTPTimeout
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if resp.StatusCode == http.StatusNotFound {
|
if resp.StatusCode == http.StatusNotFound {
|
||||||
errc <- lease.ErrLeaseNotFound
|
return nil, lease.ErrLeaseNotFound
|
||||||
return
|
|
||||||
}
|
}
|
||||||
if resp.StatusCode != http.StatusOK {
|
if resp.StatusCode != http.StatusOK {
|
||||||
errc <- fmt.Errorf("lease: unknown error(%s)", string(b))
|
return nil, fmt.Errorf("lease: unknown error(%s)", string(b))
|
||||||
return
|
|
||||||
}
|
|
||||||
errc <- nil
|
|
||||||
}()
|
|
||||||
select {
|
|
||||||
case derr := <-errc:
|
|
||||||
if derr != nil {
|
|
||||||
return nil, derr
|
|
||||||
}
|
|
||||||
case <-ctx.Done():
|
|
||||||
cancel()
|
|
||||||
return nil, ctx.Err()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
lresp := &leasepb.LeaseInternalResponse{}
|
lresp := &leasepb.LeaseInternalResponse{}
|
||||||
|
|
|
@ -13,15 +13,6 @@ import (
|
||||||
"net/http"
|
"net/http"
|
||||||
)
|
)
|
||||||
|
|
||||||
func RequestCanceler(req *http.Request) func() {
|
|
||||||
ch := make(chan struct{})
|
|
||||||
req.Cancel = ch
|
|
||||||
|
|
||||||
return func() {
|
|
||||||
close(ch)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
// GracefulClose drains http.Response.Body until it hits EOF
|
// GracefulClose drains http.Response.Body until it hits EOF
|
||||||
// and closes it. This prevents TCP/TLS connections from closing,
|
// and closes it. This prevents TCP/TLS connections from closing,
|
||||||
// therefore available for reuse.
|
// therefore available for reuse.
|
||||||
|
|
|
@ -16,6 +16,7 @@ package httpproxy
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"bytes"
|
"bytes"
|
||||||
|
"context"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
|
@ -24,11 +25,9 @@ import (
|
||||||
"net/url"
|
"net/url"
|
||||||
"strings"
|
"strings"
|
||||||
"sync/atomic"
|
"sync/atomic"
|
||||||
|
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
|
"github.com/coreos/etcd/etcdserver/api/v2http/httptypes"
|
||||||
"github.com/coreos/etcd/pkg/httputil"
|
|
||||||
"github.com/coreos/pkg/capnslog"
|
"github.com/coreos/pkg/capnslog"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -110,7 +109,9 @@ func (p *reverseProxy) ServeHTTP(rw http.ResponseWriter, clientreq *http.Request
|
||||||
var requestClosed int32
|
var requestClosed int32
|
||||||
completeCh := make(chan bool, 1)
|
completeCh := make(chan bool, 1)
|
||||||
closeNotifier, ok := rw.(http.CloseNotifier)
|
closeNotifier, ok := rw.(http.CloseNotifier)
|
||||||
cancel := httputil.RequestCanceler(proxyreq)
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
proxyreq = proxyreq.WithContext(ctx)
|
||||||
|
defer cancel()
|
||||||
if ok {
|
if ok {
|
||||||
closeCh := closeNotifier.CloseNotify()
|
closeCh := closeNotifier.CloseNotify()
|
||||||
go func() {
|
go func() {
|
||||||
|
@ -118,7 +119,6 @@ func (p *reverseProxy) ServeHTTP(rw http.ResponseWriter, clientreq *http.Request
|
||||||
case <-closeCh:
|
case <-closeCh:
|
||||||
atomic.StoreInt32(&requestClosed, 1)
|
atomic.StoreInt32(&requestClosed, 1)
|
||||||
plog.Printf("client %v closed request prematurely", clientreq.RemoteAddr)
|
plog.Printf("client %v closed request prematurely", clientreq.RemoteAddr)
|
||||||
cancel()
|
|
||||||
case <-completeCh:
|
case <-completeCh:
|
||||||
}
|
}
|
||||||
}()
|
}()
|
||||||
|
|
Loading…
Reference in New Issue