fix for empty body codec test

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

broker/http/http_test.go

@@ -8,6 +8,7 @@ import (
 
 	"github.com/google/uuid"
 	http "github.com/unistack-org/micro-broker-http"
+	jsoncodec "github.com/unistack-org/micro-codec-json"
 	rmemory "github.com/unistack-org/micro-registry-memory"
 	"github.com/unistack-org/micro/v3/broker"
 	"github.com/unistack-org/micro/v3/registry"
@@ -63,7 +64,7 @@ func sub(be *testing.B, c int) {
 	be.StopTimer()
 	m := newTestRegistry()
 
-	b := http.NewBroker(broker.Registry(m))
+	b := http.NewBroker(broker.Codec(jsoncodec.NewCodec()), broker.Registry(m))
 	topic := uuid.New().String()
 
 	if err := b.Init(); err != nil {
@@ -122,7 +123,7 @@ func sub(be *testing.B, c int) {
 func pub(be *testing.B, c int) {
 	be.StopTimer()
 	m := newTestRegistry()
-	b := http.NewBroker(broker.Registry(m))
+	b := http.NewBroker(broker.Codec(jsoncodec.NewCodec()), broker.Registry(m))
 	topic := uuid.New().String()
 
 	if err := b.Init(); err != nil {
@@ -191,7 +192,7 @@ func pub(be *testing.B, c int) {
 
 func TestBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := http.NewBroker(broker.Registry(m))
+	b := http.NewBroker(broker.Codec(jsoncodec.NewCodec()), broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)
@@ -238,7 +239,7 @@ func TestBroker(t *testing.T) {
 
 func TestConcurrentSubBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := http.NewBroker(broker.Registry(m))
+	b := http.NewBroker(broker.Codec(jsoncodec.NewCodec()), broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)
@@ -295,7 +296,7 @@ func TestConcurrentSubBroker(t *testing.T) {
 
 func TestConcurrentPubBroker(t *testing.T) {
 	m := newTestRegistry()
-	b := http.NewBroker(broker.Registry(m))
+	b := http.NewBroker(broker.Codec(jsoncodec.NewCodec()), broker.Registry(m))
 
 	if err := b.Init(); err != nil {
 		t.Fatalf("Unexpected init error: %v", err)

client/grpc/grpc_test.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	grpc "github.com/unistack-org/micro-client-grpc"
+	protocodec "github.com/unistack-org/micro-codec-proto"
 	rmemory "github.com/unistack-org/micro-registry-memory"
 	regRouter "github.com/unistack-org/micro-router-registry"
 	pb "github.com/unistack-org/micro-tests/client/grpc/proto"
@@ -94,7 +95,7 @@ func TestGRPCClient(t *testing.T) {
 	rtr := regRouter.NewRouter(router.Registry(r))
 
 	// create client
-	c := grpc.NewClient(client.Router(rtr))
+	c := grpc.NewClient(client.Codec("application/grpc+proto", protocodec.NewCodec()), client.Router(rtr))
 
 	testMethods := []string{
 		"/helloworld.Test/Call",

client/http/http_test.go

@@ -0,0 +1,290 @@
+package http
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"testing"
+
+	mhttp "github.com/unistack-org/micro-client-http"
+	jsoncodec "github.com/unistack-org/micro-codec-json"
+	rmemory "github.com/unistack-org/micro-registry-memory"
+	rrouter "github.com/unistack-org/micro-router-registry"
+	"github.com/unistack-org/micro/v3/client"
+	"github.com/unistack-org/micro/v3/codec"
+	"github.com/unistack-org/micro/v3/registry"
+	"github.com/unistack-org/micro/v3/router"
+)
+
+var (
+	defaultHTTPCodecs = map[string]codec.Codec{
+		"application/json": jsoncodec.NewCodec(),
+	}
+)
+
+type Message struct {
+	Seq  int64
+	Data string
+}
+
+func TestHTTPClient(t *testing.T) {
+	reg := rmemory.NewRegistry()
+	rtr := rrouter.NewRouter(router.Registry(reg))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/foo/bar", func(w http.ResponseWriter, r *http.Request) {
+		// only accept post
+		if r.Method != "POST" {
+			http.Error(w, "expect post method", 500)
+			return
+		}
+
+		// get codec
+		ct := r.Header.Get("Content-Type")
+		codec, ok := defaultHTTPCodecs[ct]
+		if !ok {
+			http.Error(w, "codec not found", 500)
+			return
+		}
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// extract message
+		msg := &Message{}
+		if err := codec.Unmarshal(b, msg); err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// marshal response
+		b, err = codec.Marshal(msg)
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// write response
+		w.Write(b)
+	})
+	go http.Serve(l, mux)
+
+	if err := reg.Register(ctx, &registry.Service{
+		Name: "test.service",
+		Nodes: []*registry.Node{
+			{
+				Id:      "test.service.1",
+				Address: l.Addr().String(),
+				Metadata: map[string]string{
+					"protocol": "http",
+				},
+			},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	c := mhttp.NewClient(client.ContentType("application/json"), client.Codec("application/json", jsoncodec.NewCodec()), client.Router(rtr))
+
+	for i := 0; i < 10; i++ {
+		msg := &Message{
+			Seq:  int64(i),
+			Data: fmt.Sprintf("message %d", i),
+		}
+		req := c.NewRequest("test.service", "/foo/bar", msg)
+		rsp := &Message{}
+		err := c.Call(context.TODO(), req, rsp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if rsp.Seq != msg.Seq {
+			t.Fatalf("invalid seq %d for %d", rsp.Seq, msg.Seq)
+		}
+	}
+}
+
+func TestHTTPClientStream(t *testing.T) {
+	reg := rmemory.NewRegistry()
+	rtr := rrouter.NewRouter(router.Registry(reg))
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	l, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer l.Close()
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/foo/bar", func(w http.ResponseWriter, r *http.Request) {
+		// only accept post
+		if r.Method != "POST" {
+			http.Error(w, "expect post method", 500)
+			return
+		}
+
+		// hijack the connection
+		hj, ok := w.(http.Hijacker)
+		if !ok {
+			http.Error(w, "could not hijack conn", 500)
+			return
+
+		}
+
+		// hijacked
+		conn, bufrw, err := hj.Hijack()
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+		defer conn.Close()
+
+		// read off the first request
+		// get codec
+		ct := r.Header.Get("Content-Type")
+		codec, ok := defaultHTTPCodecs[ct]
+		if !ok {
+			http.Error(w, "codec not found", 500)
+			return
+		}
+		b, err := ioutil.ReadAll(r.Body)
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// extract message
+		msg := &Message{}
+		if err := codec.Unmarshal(b, msg); err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// marshal response
+		b, err = codec.Marshal(msg)
+		if err != nil {
+			http.Error(w, err.Error(), 500)
+			return
+		}
+
+		// write response
+		rsp := &http.Response{
+			Header:        r.Header,
+			Body:          ioutil.NopCloser(bytes.NewBuffer(b)),
+			Status:        "200 OK",
+			StatusCode:    200,
+			Proto:         "HTTP/1.1",
+			ProtoMajor:    1,
+			ProtoMinor:    1,
+			ContentLength: int64(len(b)),
+		}
+
+		// write response
+		rsp.Write(bufrw)
+		bufrw.Flush()
+
+		reader := bufio.NewReader(conn)
+
+		for {
+			r, err := http.ReadRequest(reader)
+			if err != nil {
+				http.Error(w, err.Error(), 500)
+				return
+			}
+
+			b, err = ioutil.ReadAll(r.Body)
+			if err != nil {
+				http.Error(w, err.Error(), 500)
+				return
+			}
+
+			// extract message
+			msg := &Message{}
+			if err := codec.Unmarshal(b, msg); err != nil {
+				http.Error(w, err.Error(), 500)
+				return
+			}
+
+			// marshal response
+			b, err = codec.Marshal(msg)
+			if err != nil {
+				http.Error(w, err.Error(), 500)
+				return
+			}
+
+			rsp := &http.Response{
+				Header:        r.Header,
+				Body:          ioutil.NopCloser(bytes.NewBuffer(b)),
+				Status:        "200 OK",
+				StatusCode:    200,
+				Proto:         "HTTP/1.1",
+				ProtoMajor:    1,
+				ProtoMinor:    1,
+				ContentLength: int64(len(b)),
+			}
+
+			// write response
+			rsp.Write(bufrw)
+			bufrw.Flush()
+		}
+	})
+	go http.Serve(l, mux)
+
+	if err := reg.Register(ctx, &registry.Service{
+		Name: "test.service",
+		Nodes: []*registry.Node{
+			{
+				Id:      "test.service.1",
+				Address: l.Addr().String(),
+				Metadata: map[string]string{
+					"protocol": "http",
+				},
+			},
+		},
+	}); err != nil {
+		t.Fatal(err)
+	}
+
+	c := mhttp.NewClient(client.ContentType("application/json"), client.Codec("application/json", jsoncodec.NewCodec()), client.Router(rtr))
+	req := c.NewRequest("test.service", "/foo/bar", &Message{})
+	stream, err := c.Stream(context.TODO(), req)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer stream.Close()
+
+	for i := 0; i < 10; i++ {
+		msg := &Message{
+			Seq:  int64(i),
+			Data: fmt.Sprintf("message %d", i),
+		}
+		err := stream.Send(msg)
+		if err != nil {
+			t.Fatal(err)
+		}
+		rsp := &Message{}
+		err = stream.Recv(rsp)
+		if err != nil {
+			t.Fatal(err)
+		}
+		if rsp.Seq != msg.Seq {
+			t.Fatalf("invalid seq %d for %d", rsp.Seq, msg.Seq)
+		}
+	}
+}

codec/codec_test.go

@@ -0,0 +1,46 @@
+package codec_test
+
+import (
+	"testing"
+
+	grpc "github.com/unistack-org/micro-codec-grpc"
+	json "github.com/unistack-org/micro-codec-json"
+	proto "github.com/unistack-org/micro-codec-proto"
+	"github.com/unistack-org/micro/v3/codec"
+)
+
+type testRWC struct{}
+
+func (rwc *testRWC) Read(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (rwc *testRWC) Write(p []byte) (n int, err error) {
+	return 0, nil
+}
+
+func (rwc *testRWC) Close() error {
+	return nil
+}
+
+func getCodecs() map[string]codec.Codec {
+	return map[string]codec.Codec{
+		"bytes": codec.NewCodec(),
+		"grpc":  grpc.NewCodec(),
+		"json":  json.NewCodec(),
+		"proto": proto.NewCodec(),
+	}
+}
+
+func Test_WriteEmptyBody(t *testing.T) {
+	rw := &testRWC{}
+	for name, c := range getCodecs() {
+		err := c.Write(rw, &codec.Message{
+			Type:   codec.Error,
+			Header: map[string]string{},
+		}, nil)
+		if err != nil {
+			t.Fatalf("codec %s - expected no error when writing empty/nil body: %s", name, err)
+		}
+	}
+}

server/grpc/grpc_test.go

@@ -6,6 +6,7 @@ import (
 
 	bmemory "github.com/unistack-org/micro-broker-memory"
 	gclient "github.com/unistack-org/micro-client-grpc"
+	protocodec "github.com/unistack-org/micro-codec-proto"
 	rmemory "github.com/unistack-org/micro-registry-memory"
 	regRouter "github.com/unistack-org/micro-router-registry"
 	gserver "github.com/unistack-org/micro-server-grpc"
@@ -34,7 +35,7 @@ func TestGRPCServer(t *testing.T) {
 
 	r := rmemory.NewRegistry()
 	b := bmemory.NewBroker(broker.Registry(r))
-	s := gserver.NewServer(server.Address(":12345"), server.Registry(r), server.Name("helloworld"), gserver.Reflection(true))
+	s := gserver.NewServer(server.Codec("application/grpc+proto", protocodec.NewCodec()), server.Address(":12345"), server.Registry(r), server.Name("helloworld"), gserver.Reflection(true))
 	// create router
 	rtr := regRouter.NewRouter(router.Registry(r))
 
@@ -59,7 +60,7 @@ func TestGRPCServer(t *testing.T) {
 	}()
 
 	// create client
-	c := gclient.NewClient(client.Router(rtr), client.Registry(r), client.Broker(b))
+	c := gclient.NewClient(client.Codec("application/grpc+proto", protocodec.NewCodec()), client.Router(rtr), client.Registry(r), client.Broker(b))
 
 	testMethods := []string{
 		"Test.Call",

wrapper/metrics/prometheus/prometheus_test.go

@@ -0,0 +1,189 @@
+package prometheus_test
+
+import (
+	"context"
+	"fmt"
+	"testing"
+
+	"github.com/prometheus/client_golang/prometheus"
+	dto "github.com/prometheus/client_model/go"
+	"github.com/stretchr/testify/assert"
+	bmemory "github.com/unistack-org/micro-broker-memory"
+	cli "github.com/unistack-org/micro-client-grpc"
+	jsoncodec "github.com/unistack-org/micro-codec-json"
+	promwrapper "github.com/unistack-org/micro-metrics-prometheus"
+	rmemory "github.com/unistack-org/micro-registry-memory"
+	rrouter "github.com/unistack-org/micro-router-registry"
+	srv "github.com/unistack-org/micro-server-grpc"
+	"github.com/unistack-org/micro/v3/broker"
+	"github.com/unistack-org/micro/v3/client"
+	"github.com/unistack-org/micro/v3/router"
+	"github.com/unistack-org/micro/v3/server"
+)
+
+type Test interface {
+	Method(ctx context.Context, in *TestRequest, opts ...client.CallOption) (*TestResponse, error)
+}
+
+type TestRequest struct {
+	IsError bool
+}
+type TestResponse struct{}
+
+type testHandler struct{}
+
+func (t *testHandler) Method(ctx context.Context, req *TestRequest, rsp *TestResponse) error {
+	if req.IsError {
+		return fmt.Errorf("test error")
+	}
+	return nil
+}
+
+func TestPrometheusMetrics(t *testing.T) {
+	client.DefaultRetries = 0
+	// setup
+	reg := rmemory.NewRegistry()
+	brk := bmemory.NewBroker(broker.Registry(reg))
+
+	name := "test"
+	id := "id-1234567890"
+	version := "1.2.3.4"
+	rt := rrouter.NewRouter(router.Registry(reg))
+
+	c := cli.NewClient(
+		client.Codec("application/grpc+json", jsoncodec.NewCodec()),
+		client.Codec("application/json", jsoncodec.NewCodec()),
+		client.Router(rt),
+	)
+	s := srv.NewServer(
+		server.Codec("application/grpc+json", jsoncodec.NewCodec()),
+		server.Codec("application/json", jsoncodec.NewCodec()),
+		server.Name(name),
+		server.Version(version),
+		server.Id(id),
+		server.Registry(reg),
+		server.Broker(brk),
+		server.WrapHandler(
+			promwrapper.NewHandlerWrapper(
+				promwrapper.ServiceName(name),
+				promwrapper.ServiceVersion(version),
+				promwrapper.ServiceID(id),
+			),
+		),
+	)
+
+	if err := s.Init(); err != nil {
+		t.Fatal(err)
+	}
+	type Test struct {
+		*testHandler
+	}
+
+	s.Handle(
+		s.NewHandler(&Test{new(testHandler)}),
+	)
+
+	if err := s.Start(); err != nil {
+		t.Fatalf("Unexpected error starting server: %v", err)
+	}
+	defer s.Stop()
+
+	req := c.NewRequest(name, "Test.Method", &TestRequest{IsError: false}, client.WithContentType("application/json"))
+	rsp := TestResponse{}
+
+	assert.NoError(t, c.Call(context.TODO(), req, &rsp))
+
+	req = c.NewRequest(name, "Test.Method", &TestRequest{IsError: true}, client.WithContentType("application/json"))
+	assert.Error(t, c.Call(context.TODO(), req, &rsp))
+
+	list, _ := prometheus.DefaultGatherer.Gather()
+
+	metric := findMetricByName(list, dto.MetricType_SUMMARY, "micro_server_latency_microseconds")
+
+	if metric == nil || metric.Metric == nil || len(metric.Metric) == 0 {
+		t.Fatalf("no metrics returned")
+	}
+
+	for _, v := range metric.Metric[0].Label {
+		switch *v.Name {
+		case "micro_version":
+			assert.Equal(t, version, *v.Value)
+		case "micro_id":
+			assert.Equal(t, id, *v.Value)
+		case "micro_name":
+			assert.Equal(t, name, *v.Value)
+		case "micro_endpoint":
+			assert.Equal(t, "Test.Method", *v.Value)
+		default:
+			t.Fatalf("unknown %v with %v", *v.Name, *v.Value)
+		}
+	}
+
+	assert.Equal(t, uint64(2), *metric.Metric[0].Summary.SampleCount)
+	assert.True(t, *metric.Metric[0].Summary.SampleSum > 0)
+
+	metric = findMetricByName(list, dto.MetricType_HISTOGRAM, "micro_server_request_duration_seconds")
+
+	for _, v := range metric.Metric[0].Label {
+		switch *v.Name {
+		case "micro_version":
+			assert.Equal(t, version, *v.Value)
+		case "micro_id":
+			assert.Equal(t, id, *v.Value)
+		case "micro_name":
+			assert.Equal(t, name, *v.Value)
+		case "micro_endpoint":
+			assert.Equal(t, "Test.Method", *v.Value)
+		default:
+			t.Fatalf("unknown %v with %v", *v.Name, *v.Value)
+		}
+	}
+
+	assert.Equal(t, uint64(2), *metric.Metric[0].Histogram.SampleCount)
+	assert.True(t, *metric.Metric[0].Histogram.SampleSum > 0)
+
+	metric = findMetricByName(list, dto.MetricType_COUNTER, "micro_server_request_total")
+
+	for _, v := range metric.Metric[0].Label {
+		switch *v.Name {
+		case "micro_version":
+			assert.Equal(t, version, *v.Value)
+		case "micro_id":
+			assert.Equal(t, id, *v.Value)
+		case "micro_name":
+			assert.Equal(t, name, *v.Value)
+		case "micro_endpoint":
+			assert.Equal(t, "Test.Method", *v.Value)
+		case "micro_status":
+			assert.Equal(t, "failure", *v.Value)
+		}
+	}
+	assert.Equal(t, *metric.Metric[0].Counter.Value, float64(1))
+
+	for _, v := range metric.Metric[1].Label {
+		switch *v.Name {
+		case "micro_version":
+			assert.Equal(t, version, *v.Value)
+		case "micro_id":
+			assert.Equal(t, id, *v.Value)
+		case "micro_name":
+			assert.Equal(t, name, *v.Value)
+		case "micro_endpoint":
+			assert.Equal(t, "Test.Method", *v.Value)
+		case "micro_status":
+			assert.Equal(t, "success", *v.Value)
+		}
+	}
+
+	assert.Equal(t, *metric.Metric[1].Counter.Value, float64(1))
+}
+
+func findMetricByName(list []*dto.MetricFamily, tp dto.MetricType, name string) *dto.MetricFamily {
+	for _, metric := range list {
+		if *metric.Name == name && *metric.Type == tp {
+			return metric
+		}
+	}
+
+	return nil
+}

wrapper/opentracing/opentracing_test.go

@@ -10,6 +10,7 @@ import (
 	"github.com/stretchr/testify/assert"
 	rbroker "github.com/unistack-org/micro-broker-memory"
 	cli "github.com/unistack-org/micro-client-grpc"
+	jsoncodec "github.com/unistack-org/micro-codec-json"
 	rmemory "github.com/unistack-org/micro-registry-memory"
 	rrouter "github.com/unistack-org/micro-router-registry"
 	srv "github.com/unistack-org/micro-server-grpc"
@@ -79,11 +80,15 @@ func TestClient(t *testing.T) {
 			rt := rrouter.NewRouter(router.Registry(reg))
 
 			c := cli.NewClient(
+				client.Codec("application/grpc+json", jsoncodec.NewCodec()),
+				client.Codec("application/json", jsoncodec.NewCodec()),
 				client.Router(rt),
 				client.WrapCall(otwrapper.NewClientCallWrapper(otwrapper.WithTracer(tracer))),
 			)
 
 			s := srv.NewServer(
+				server.Codec("application/grpc+json", jsoncodec.NewCodec()),
+				server.Codec("application/json", jsoncodec.NewCodec()),
 				server.Name(serverName),
 				server.Version(serverVersion),
 				server.Id(serverID),