|
@@ -44,6 +44,7 @@ type clientConn struct {
|
|
|
nextRes *http.Response
|
|
nextRes *http.Response
|
|
|
|
|
|
|
|
mu sync.Mutex
|
|
mu sync.Mutex
|
|
|
|
|
+ closed bool
|
|
|
goAway *GoAwayFrame // if non-nil, the GoAwayFrame we received
|
|
goAway *GoAwayFrame // if non-nil, the GoAwayFrame we received
|
|
|
streams map[uint32]*clientStream
|
|
streams map[uint32]*clientStream
|
|
|
nextStreamID uint32
|
|
nextStreamID uint32
|
|
@@ -100,7 +101,7 @@ func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
|
|
|
return nil, err
|
|
return nil, err
|
|
|
}
|
|
}
|
|
|
res, err := cc.roundTrip(req)
|
|
res, err := cc.roundTrip(req)
|
|
|
- if isShutdownError(err) { // TODO: or clientconn is overloaded (too many outstanding requests)?
|
|
|
|
|
|
|
+ if shouldRetryRequest(err) { // TODO: or clientconn is overloaded (too many outstanding requests)?
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
if err != nil {
|
|
if err != nil {
|
|
@@ -123,9 +124,11 @@ func (t *Transport) CloseIdleConnections() {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
-func isShutdownError(err error) bool {
|
|
|
|
|
- // TODO: implement
|
|
|
|
|
- return false
|
|
|
|
|
|
|
+var errClientConnClosed = errors.New("http2: client conn is closed")
|
|
|
|
|
+
|
|
|
|
|
+func shouldRetryRequest(err error) bool {
|
|
|
|
|
+ // TODO: or GOAWAY graceful shutdown stuff
|
|
|
|
|
+ return err == errClientConnClosed
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (t *Transport) removeClientConn(cc *clientConn) {
|
|
func (t *Transport) removeClientConn(cc *clientConn) {
|
|
@@ -136,7 +139,12 @@ func (t *Transport) removeClientConn(cc *clientConn) {
|
|
|
if !ok {
|
|
if !ok {
|
|
|
continue
|
|
continue
|
|
|
}
|
|
}
|
|
|
- t.conns[key] = filterOutClientConn(vv, cc)
|
|
|
|
|
|
|
+ newList := filterOutClientConn(vv, cc)
|
|
|
|
|
+ if len(newList) > 0 {
|
|
|
|
|
+ t.conns[key] = newList
|
|
|
|
|
+ } else {
|
|
|
|
|
+ delete(t.conns, key)
|
|
|
|
|
+ }
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
@@ -269,20 +277,32 @@ func (cc *clientConn) setGoAway(f *GoAwayFrame) {
|
|
|
func (cc *clientConn) canTakeNewRequest() bool {
|
|
func (cc *clientConn) canTakeNewRequest() bool {
|
|
|
cc.mu.Lock()
|
|
cc.mu.Lock()
|
|
|
defer cc.mu.Unlock()
|
|
defer cc.mu.Unlock()
|
|
|
- return cc.goAway == nil && int64(len(cc.streams)+1) < int64(cc.maxConcurrentStreams)
|
|
|
|
|
|
|
+ return cc.goAway == nil &&
|
|
|
|
|
+ int64(len(cc.streams)+1) < int64(cc.maxConcurrentStreams) &&
|
|
|
|
|
+ cc.nextStreamID < 2147483647
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (cc *clientConn) closeIfIdle() {
|
|
func (cc *clientConn) closeIfIdle() {
|
|
|
cc.mu.Lock()
|
|
cc.mu.Lock()
|
|
|
- defer cc.mu.Unlock()
|
|
|
|
|
if len(cc.streams) > 0 {
|
|
if len(cc.streams) > 0 {
|
|
|
|
|
+ cc.mu.Unlock()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
|
|
+ cc.closed = true
|
|
|
|
|
+ // TODO: do clients send GOAWAY too? maybe? Just Close:
|
|
|
|
|
+ cc.mu.Unlock()
|
|
|
|
|
+
|
|
|
|
|
+ cc.tconn.Close()
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
func (cc *clientConn) roundTrip(req *http.Request) (*http.Response, error) {
|
|
func (cc *clientConn) roundTrip(req *http.Request) (*http.Response, error) {
|
|
|
cc.mu.Lock()
|
|
cc.mu.Lock()
|
|
|
|
|
|
|
|
|
|
+ if cc.closed {
|
|
|
|
|
+ cc.mu.Unlock()
|
|
|
|
|
+ return nil, errClientConnClosed
|
|
|
|
|
+ }
|
|
|
|
|
+
|
|
|
cs := cc.newStream()
|
|
cs := cc.newStream()
|
|
|
hasBody := false // TODO
|
|
hasBody := false // TODO
|
|
|
|
|
|
|
@@ -413,8 +433,6 @@ func (cc *clientConn) readLoop() {
|
|
|
}
|
|
}
|
|
|
}()
|
|
}()
|
|
|
|
|
|
|
|
- defer println("Transport readLoop returning")
|
|
|
|
|
-
|
|
|
|
|
// continueStreamID is the stream ID we're waiting for
|
|
// continueStreamID is the stream ID we're waiting for
|
|
|
// continuation frames for.
|
|
// continuation frames for.
|
|
|
var continueStreamID uint32
|
|
var continueStreamID uint32
|