move all the buffer references to util/buf

client/grpc/buffer.go

@@ -1,14 +0,0 @@
-package grpc
-
-import (
-	"bytes"
-)
-
-type buffer struct {
-	*bytes.Buffer
-}
-
-func (b *buffer) Close() error {
-	b.Buffer.Reset()
-	return nil
-}

client/grpc/grpc.go

@@ -2,7 +2,6 @@
 package grpc
 
 import (
-	"bytes"
 	"context"
 	"crypto/tls"
 	"fmt"
@@ -19,6 +18,7 @@ import (
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/transport"
 
+	"github.com/micro/go-micro/util/buf"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials"
 	"google.golang.org/grpc/encoding"
@@ -491,7 +491,8 @@ func (g *grpcClient) Publish(ctx context.Context, p client.Message, opts ...clie
 		return errors.InternalServerError("go.micro.client", err.Error())
 	}
 
-	b := &buffer{bytes.NewBuffer(nil)}
+	b := buf.New(nil)
+
 	if err := cf(b).Write(&codec.Message{Type: codec.Event}, p.Payload()); err != nil {
 		return errors.InternalServerError("go.micro.client", err.Error())
 	}

client/rpc_client.go

@@ -1,7 +1,6 @@
 package client
 
 import (
-	"bytes"
 	"context"
 	"fmt"
 	"os"
@@ -18,6 +17,7 @@ import (
 	"github.com/micro/go-micro/metadata"
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/transport"
+	"github.com/micro/go-micro/util/buf"
 )
 
 type rpcClient struct {
@@ -538,7 +538,10 @@ func (r *rpcClient) Publish(ctx context.Context, msg Message, opts ...PublishOpt
 	if err != nil {
 		return errors.InternalServerError("go.micro.client", err.Error())
 	}
-	b := &buffer{bytes.NewBuffer(nil)}
+
+	// new buffer
+	b := buf.New(nil)
+
 	if err := cf(b).Write(&codec.Message{
 		Target: topic,
 		Type:   codec.Event,

server/buffer.go

@@ -1,14 +0,0 @@
-package server
-
-import (
-	"bytes"
-)
-
-type buffer struct {
-	*bytes.Buffer
-}
-
-func (b *buffer) Close() error {
-	b.Buffer.Reset()
-	return nil
-}

server/grpc/buffer.go

@@ -1,14 +0,0 @@
-package grpc
-
-import (
-	"bytes"
-)
-
-type buffer struct {
-	*bytes.Buffer
-}
-
-func (b *buffer) Close() error {
-	b.Buffer.Reset()
-	return nil
-}

server/grpc/subscriber.go

@@ -12,6 +12,7 @@ import (
 	"github.com/micro/go-micro/metadata"
 	"github.com/micro/go-micro/registry"
 	"github.com/micro/go-micro/server"
+	"github.com/micro/go-micro/util/buf"
 )
 
 const (
@@ -204,7 +205,7 @@ func (g *grpcServer) createSubHandler(sb *subscriber, opts server.Options) broke
 				req = req.Elem()
 			}
 
-			b := &buffer{bytes.NewBuffer(msg.Body)}
+			b := buf.New(bytes.NewBuffer(msg.Body))
 			co := cf(b)
 			defer co.Close()
 

server/subscriber.go

@@ -11,6 +11,7 @@ import (
 	"github.com/micro/go-micro/codec"
 	"github.com/micro/go-micro/metadata"
 	"github.com/micro/go-micro/registry"
+	"github.com/micro/go-micro/util/buf"
 )
 
 const (
@@ -210,7 +211,7 @@ func (s *rpcServer) createSubHandler(sb *subscriber, opts Options) broker.Handle
 				req = req.Elem()
 			}
 
-			b := &buffer{bytes.NewBuffer(msg.Body)}
+			b := buf.New(bytes.NewBuffer(msg.Body))
 			co := cf(b)
 			defer co.Close()
 

transport/http_transport.go

@@ -14,16 +14,13 @@ import (
 	"time"
 
 	maddr "github.com/micro/go-micro/util/addr"
+	"github.com/micro/go-micro/util/buf"
 	mnet "github.com/micro/go-micro/util/net"
 	mls "github.com/micro/go-micro/util/tls"
 	"golang.org/x/net/http2"
 	"golang.org/x/net/http2/h2c"
 )
 
-type buffer struct {
-	io.ReadWriter
-}
-
 type httpTransport struct {
 	opts Options
 }
@@ -65,10 +62,6 @@ type httpTransportListener struct {
 	listener net.Listener
 }
 
-func (b *buffer) Close() error {
-	return nil
-}
-
 func (h *httpTransportClient) Local() string {
 	return h.local
 }
@@ -84,11 +77,8 @@ func (h *httpTransportClient) Send(m *Message) error {
 		header.Set(k, v)
 	}
 
-	reqB := bytes.NewBuffer(m.Body)
+	b := buf.New(bytes.NewBuffer(m.Body))
-	defer reqB.Reset()
+	defer b.Close()
-	buf := &buffer{
-		reqB,
-	}
 
 	req := &http.Request{
 		Method: "POST",
@@ -97,8 +87,8 @@ func (h *httpTransportClient) Send(m *Message) error {
 			Host:   h.addr,
 		},
 		Header:        header,
-		Body:          buf,
+		Body:          b,
-		ContentLength: int64(reqB.Len()),
+		ContentLength: int64(b.Len()),
 		Host:          h.addr,
 	}
 

util/buf/buf.go

@@ -1,4 +1,4 @@
-package client
+package buf
 
 import (
 	"bytes"
@@ -12,3 +12,10 @@ func (b *buffer) Close() error {
 	b.Buffer.Reset()
 	return nil
 }
+
+func New(b *bytes.Buffer) *buffer {
+	if b == nil {
+		b = bytes.NewBuffer(nil)
+	}
+	return &buffer{b}
+}