WIP(x/net/http/client): Implement BodyChunk

x/net/http/client.go

@@ -307,7 +307,7 @@ func send(ireq *Request, rt RoundTripper, deadline time.Time) (resp *Response, d
 		forkReq()
 	}
 
-	// TODO(spongehah) timeout(send)
+	// TODO(spongehah) tmp timeout(send)
 	//stopTimer, didTimeout := setRequestCancel(req, rt, deadline)
 	req.timeoutch = make(chan struct{}, 1)
 	req.deadline = deadline

x/net/http/request.go

@@ -294,7 +294,7 @@ func (r *Request) write(client *hyper.ClientConn, taskData *taskData, exec *hype
 	//}
 
 	// Prepare the hyper.Request
-	hyperReq, err := r.newHyperRequest(taskData.pc.isProxy, taskData.req.extra)
+	hyperReq, err := r.newHyperRequest(taskData.pc.isProxy, taskData.req.extra, taskData.req)
 	if err != nil {
 		return err
 	}
@@ -308,7 +308,7 @@ func (r *Request) write(client *hyper.ClientConn, taskData *taskData, exec *hype
 	return err
 }
 
-func (r *Request) newHyperRequest(usingProxy bool, extraHeader Header) (*hyper.Request, error) {
+func (r *Request) newHyperRequest(usingProxy bool, extraHeader Header, treq *transportRequest) (*hyper.Request, error) {
 	// Find the target host. Prefer the Host: header, but if that
 	// is not given, use the host from the request URL.
 	//
@@ -401,11 +401,6 @@ func (r *Request) newHyperRequest(usingProxy bool, extraHeader Header) (*hyper.R
 	}
 
 	// Process Body,ContentLength,Close,Trailer
-	//tw, err := newTransferWriter(r)
-	//if err != nil {
-	//	return err
-	//}
-	//err = tw.writeHeader(w, trace)
 	err = r.writeHeader(reqHeaders)
 	if err != nil {
 		return nil, err
@@ -433,7 +428,7 @@ func (r *Request) newHyperRequest(usingProxy bool, extraHeader Header) (*hyper.R
 	}
 
 	// Write body and trailer
-	err = r.writeBody(hyperReq)
+	err = r.writeBody(hyperReq, treq)
 	if err != nil {
 		return nil, err
 	}

x/net/http/response.go

@@ -32,7 +32,7 @@ func (r *Response) closeBody() {
 	}
 }
 
-func ReadResponse(r *io.PipeReader, req *Request, hyperResp *hyper.Response) (*Response, error) {
+func ReadResponse(r *bodyChunk, req *Request, hyperResp *hyper.Response) (*Response, error) {
 	resp := &Response{
 		Request: req,
 		Header:  make(Header),
@@ -117,3 +117,73 @@ func (r *Response) bodyIsWritable() bool {
 	_, ok := r.Body.(io.Writer)
 	return ok
 }
+
+func (resp *Response) checkRespBody(taskData *taskData) (needContinue bool) {
+	pc := taskData.pc
+	bodyWritable := resp.bodyIsWritable()
+	hasBody := taskData.req.Method != "HEAD" && resp.ContentLength != 0
+
+	if resp.Close || taskData.req.Close || resp.StatusCode <= 199 || bodyWritable {
+		// Don't do keep-alive on error if either party requested a close
+		// or we get an unexpected informational (1xx) response.
+		// StatusCode 100 is already handled above.
+		pc.alive = false
+	}
+
+	if !hasBody || bodyWritable {
+		replaced := pc.t.replaceReqCanceler(taskData.req.cancelKey, nil)
+
+		// Put the idle conn back into the pool before we send the response
+		// so if they process it quickly and make another request, they'll
+		// get this same conn. But we use the unbuffered channel 'rc'
+		// to guarantee that persistConn.roundTrip got out of its select
+		// potentially waiting for this persistConn to close.
+		pc.alive = pc.alive &&
+			replaced && pc.tryPutIdleConn()
+
+		if bodyWritable {
+			pc.closeErr = errCallerOwnsConn
+		}
+
+		select {
+		case taskData.resc <- responseAndError{res: resp}:
+		case <-taskData.callerGone:
+			readLoopDefer(pc, true)
+			return true
+		}
+		// Now that they've read from the unbuffered channel, they're safely
+		// out of the select that also waits on this goroutine to die, so
+		// we're allowed to exit now if needed (if alive is false)
+		readLoopDefer(pc, false)
+		return true
+	}
+	return false
+}
+
+func (r *Response) wrapBodyEOFSignalAndGzip(taskData *taskData) {
+	body := &bodyEOFSignal{
+		body: r.Body,
+		earlyCloseFn: func() error {
+			return nil
+		},
+		fn: func(err error) error {
+			isEOF := err == io.EOF
+			if !isEOF {
+				if cerr := taskData.pc.canceled(); cerr != nil {
+					return cerr
+				}
+			}
+			return err
+		},
+	}
+	r.Body = body
+	// TODO(spongehah) gzip(wrapBodyEOFSignal)
+	//if taskData.addedGzip && EqualFold(r.Header.Get("Content-Encoding"), "gzip") {
+	//	println("gzip reader")
+	//	r.Body = &gzipReader{body: body}
+	//	r.Header.Del("Content-Encoding")
+	//	r.Header.Del("Content-Length")
+	//	r.ContentLength = -1
+	//	r.Uncompressed = true
+	//}
+}

x/net/http/transfer.go

@@ -98,7 +98,7 @@ func (uste *unsupportedTEError) Error() string {
 }
 
 // msg is *Request or *Response.
-func readTransfer(msg any, r *io.PipeReader) (err error) {
+func readTransfer(msg any, r *bodyChunk) (err error) {
 	t := &transferReader{RequestMethod: "GET"}
 
 	// Unify input
@@ -173,8 +173,6 @@ func readTransfer(msg any, r *io.PipeReader) (err error) {
 		if isResponse && noResponseBodyExpected(t.RequestMethod) || !bodyAllowedForStatus(t.StatusCode) {
 			t.Body = NoBody
 		} else {
-			// TODO(spongehah) ChunkReader(readTransfer)
-			//t.Body = &body{src: internal.NewChunkedReader(r), hdr: msg, r: r, closing: t.Close}
 			t.Body = &body{src: r, hdr: msg, r: r, closing: t.Close}
 		}
 	case realLength == 0:
@@ -665,15 +663,15 @@ func (req *Request) unwrapBody() io.Reader {
 	return req.Body
 }
 
-func (r *Request) writeBody(hyperReq *hyper.Request) error {
+func (r *Request) writeBody(hyperReq *hyper.Request, treq *transportRequest) error {
 	if r.Body != nil {
 		var body = r.unwrapBody()
 		hyperReqBody := hyper.NewBody()
 		buf := make([]byte, defaultChunkSize)
 		reqData := &bodyReq{
-			body:      body,
-			buf:       buf,
-			closeBody: r.closeBody,
+			body: body,
+			buf:  buf,
+			treq: treq,
 		}
 		hyperReqBody.SetUserdata(c.Pointer(reqData))
 		hyperReqBody.SetDataFunc(setPostData)
@@ -683,9 +681,9 @@ func (r *Request) writeBody(hyperReq *hyper.Request) error {
 }
 
 type bodyReq struct {
-	body      io.Reader
-	buf       []byte
-	closeBody func() error
+	body io.Reader
+	buf  []byte
+	treq *transportRequest
 }
 
 func setPostData(userdata c.Pointer, ctx *hyper.Context, chunk **hyper.Buf) c.Int {
@@ -694,10 +692,11 @@ func setPostData(userdata c.Pointer, ctx *hyper.Context, chunk **hyper.Buf) c.In
 	if err != nil {
 		if err == io.EOF {
 			*chunk = nil
-			req.closeBody()
+			req.treq.closeBody()
 			return hyper.PollReady
 		}
 		fmt.Println("error reading request body: ", err)
+		req.treq.setError(requestBodyReadError{err})
 		return hyper.PollError
 	}
 	if n > 0 {
@@ -706,10 +705,11 @@ func setPostData(userdata c.Pointer, ctx *hyper.Context, chunk **hyper.Buf) c.In
 	}
 	if n == 0 {
 		*chunk = nil
-		req.closeBody()
+		req.treq.closeBody()
 		return hyper.PollReady
 	}
-	req.closeBody()
-	fmt.Printf("error reading request body: %s\n", c.GoString(c.Strerror(os.Errno)))
+	req.treq.closeBody()
+	err = fmt.Errorf("error reading request body: %s\n", c.GoString(c.Strerror(os.Errno)))
+	req.treq.setError(requestBodyReadError{err})
 	return hyper.PollError
 }