Merge pull request 'fix request/response md handling' (#113) from reqrsp-md into v3

Reviewed-on: #113

http.go

@@ -147,6 +147,11 @@ func newRequest(ctx context.Context, log logger.Logger, addr string, req client.
 	if opts.AuthToken != "" {
 		header.Set(metadata.HeaderAuthorization, opts.AuthToken)
 	}
+	if opts.RequestMetadata != nil {
+		for k, v := range opts.RequestMetadata {
+			header.Set(k, v)
+		}
+	}
 
 	if md, ok := metadata.FromOutgoingContext(ctx); ok {
 		for k, v := range md {

stream.go

@@ -2,11 +2,9 @@ package http
 
 import (
 	"bufio"
-	"bytes"
 	"context"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net"
 	"net/http"
 	"sync"
@@ -121,59 +119,55 @@ func (h *httpStream) Close() error {
 
 func (h *httpStream) parseRsp(ctx context.Context, log logger.Logger, hrsp *http.Response, cf codec.Codec, rsp interface{}, opts client.CallOptions) error {
 	var err error
+	var buf []byte
 
-	select {
-	case <-ctx.Done():
-		err = ctx.Err()
-	default:
 	// fast path return
 	if hrsp.StatusCode == http.StatusNoContent {
 		return nil
 	}
 
-		if hrsp.StatusCode < 400 {
+	select {
-			if log.V(logger.DebugLevel) {
+	case <-ctx.Done():
-				buf, rerr := io.ReadAll(hrsp.Body)
+		err = ctx.Err()
-				log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
+	default:
+		if hrsp.Body != nil {
+			buf, err = io.ReadAll(hrsp.Body)
 			if err != nil {
-					return errors.InternalServerError("go.micro.client", rerr.Error())
+				if log.V(logger.ErrorLevel) {
+					log.Errorf(ctx, "failed to read body: %v", err)
 				}
-				hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
+				return errors.InternalServerError("go.micro.client", string(buf))
 			}
-			if err = cf.ReadBody(hrsp.Body, rsp); err != nil {
+		}
+
+		if log.V(logger.DebugLevel) {
+			log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
+		}
+
+		if hrsp.StatusCode < 400 {
+			if err = cf.Unmarshal(buf, rsp); err != nil {
 				return errors.InternalServerError("go.micro.client", err.Error())
 			}
 			return nil
 		}
 
+		var rerr interface{}
 		errmap, ok := opts.Context.Value(errorMapKey{}).(map[string]interface{})
 		if ok && errmap != nil {
-			if err, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
+			if rerr, ok = errmap[fmt.Sprintf("%d", hrsp.StatusCode)].(error); !ok {
-				err, ok = errmap["default"].(error)
+				rerr, ok = errmap["default"].(error)
 			}
 		}
-		if !ok || err == nil {
+		if !ok || rerr == nil {
-			buf, cerr := io.ReadAll(hrsp.Body)
-			if log.V(logger.DebugLevel) {
-				log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
-			}
-			if cerr != nil {
-				return errors.InternalServerError("go.micro.client", cerr.Error())
-			}
 			return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
 		}
 
-		if log.V(logger.DebugLevel) {
+		if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
-			buf, rerr := io.ReadAll(hrsp.Body)
+			return errors.InternalServerError("go.micro.client", cerr.Error())
-			log.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
-			if err != nil {
-				return errors.InternalServerError("go.micro.client", rerr.Error())
-			}
-			hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
 		}
 
-		if cerr := cf.ReadBody(hrsp.Body, err); cerr != nil {
+		if err, ok = rerr.(error); !ok {
-			err = errors.InternalServerError("go.micro.client", cerr.Error())
+			err = &Error{rerr}
 		}
 	}
 

util.go

@@ -1,11 +1,9 @@
 package http
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net/http"
 	"net/url"
 	"reflect"
@@ -15,6 +13,7 @@ import (
 	"go.unistack.org/micro/v3/client"
 	"go.unistack.org/micro/v3/errors"
 	"go.unistack.org/micro/v3/logger"
+	"go.unistack.org/micro/v3/metadata"
 	rutil "go.unistack.org/micro/v3/util/reflect"
 )
 
@@ -247,45 +246,55 @@ func newTemplate(path string) ([]string, error) {
 
 func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp interface{}, opts client.CallOptions) error {
 	var err error
+	var buf []byte
+
+	// fast path return
+	if hrsp.StatusCode == http.StatusNoContent {
+		return nil
+	}
+
+	if opts.ResponseMetadata != nil {
+		*opts.ResponseMetadata = metadata.New(len(hrsp.Header))
+		for k, v := range hrsp.Header {
+			opts.ResponseMetadata.Set(k, strings.Join(v, ","))
+		}
+	}
 
 	select {
 	case <-ctx.Done():
 		err = ctx.Err()
 	default:
-		// fast path return
-		if hrsp.StatusCode == http.StatusNoContent {
-			return nil
-		}
 		ct := DefaultContentType
 
 		if htype := hrsp.Header.Get("Content-Type"); htype != "" {
 			ct = htype
 		}
 
-		cf, cerr := h.newCodec(ct)
-		if hrsp.StatusCode >= 400 && cerr != nil {
-			var buf []byte
 		if hrsp.Body != nil {
 			buf, err = io.ReadAll(hrsp.Body)
-				if err != nil && h.opts.Logger.V(logger.ErrorLevel) {
+			if err != nil {
+				if h.opts.Logger.V(logger.ErrorLevel) {
 					h.opts.Logger.Errorf(ctx, "failed to read body: %v", err)
 				}
+				return errors.InternalServerError("go.micro.client", string(buf))
 			}
-			if h.opts.Logger.V(logger.DebugLevel) {
-				h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
 		}
-			// response like text/plain or something else, return original error
+
-			return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
+		cf, cerr := h.newCodec(ct)
-		} else if cerr != nil {
+		if cerr != nil {
 			if h.opts.Logger.V(logger.DebugLevel) {
-				h.opts.Logger.Debugf(ctx, "response with %v unknown content-type", hrsp.Header, ct)
+				h.opts.Logger.Debugf(ctx, "response with %v unknown content-type %s %s", hrsp.Header, ct, buf)
 			}
 			return errors.InternalServerError("go.micro.client", cerr.Error())
 		}
 
+		if h.opts.Logger.V(logger.DebugLevel) {
+			h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
+		}
+
 		// succeseful response
 		if hrsp.StatusCode < 400 {
-			if err = cf.ReadBody(hrsp.Body, rsp); err != nil {
+			if err = cf.Unmarshal(buf, rsp); err != nil {
 				return errors.InternalServerError("go.micro.client", err.Error())
 			}
 			return nil
@@ -302,26 +311,10 @@ func (h *httpClient) parseRsp(ctx context.Context, hrsp *http.Response, rsp inte
 		}
 
 		if !ok || rerr == nil {
-			buf, rerr := io.ReadAll(hrsp.Body)
-			if h.opts.Logger.V(logger.DebugLevel) {
-				h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
-			}
-			if rerr != nil {
-				return errors.InternalServerError("go.micro.client", rerr.Error())
-			}
 			return errors.New("go.micro.client", string(buf), int32(hrsp.StatusCode))
 		}
 
-		if h.opts.Logger.V(logger.DebugLevel) {
+		if cerr := cf.Unmarshal(buf, rerr); cerr != nil {
-			buf, rerr := io.ReadAll(hrsp.Body)
-			h.opts.Logger.Debugf(ctx, "response %s with %v", buf, hrsp.Header)
-			if err != nil {
-				return errors.InternalServerError("go.micro.client", rerr.Error())
-			}
-			hrsp.Body = ioutil.NopCloser(bytes.NewBuffer(buf))
-		}
-
-		if cerr := cf.ReadBody(hrsp.Body, rerr); cerr != nil {
 			return errors.InternalServerError("go.micro.client", cerr.Error())
 		}