From 1db98ee0f08c457c31a2c1c263d1e7cea453cec1 Mon Sep 17 00:00:00 2001 From: Asim Aslam Date: Sun, 28 Jul 2019 19:33:24 +0100 Subject: [PATCH] move all the buffer references to util/buf --- client/grpc/buffer.go | 14 -------------- client/grpc/grpc.go | 5 +++-- client/rpc_client.go | 7 +++++-- server/buffer.go | 14 -------------- server/grpc/buffer.go | 14 -------------- server/grpc/subscriber.go | 3 ++- server/subscriber.go | 3 ++- transport/http_transport.go | 20 +++++--------------- client/buffer.go => util/buf/buf.go | 9 ++++++++- 9 files changed, 25 insertions(+), 64 deletions(-) delete mode 100644 client/grpc/buffer.go delete mode 100644 server/buffer.go delete mode 100644 server/grpc/buffer.go rename client/buffer.go => util/buf/buf.go (52%) diff --git a/client/grpc/buffer.go b/client/grpc/buffer.go deleted file mode 100644 index c43bb231..00000000 --- a/client/grpc/buffer.go +++ /dev/null @@ -1,14 +0,0 @@ -package grpc - -import ( - "bytes" -) - -type buffer struct { - *bytes.Buffer -} - -func (b *buffer) Close() error { - b.Buffer.Reset() - return nil -} diff --git a/client/grpc/grpc.go b/client/grpc/grpc.go index 6f5c3789..c6a3f5dd 100644 --- a/client/grpc/grpc.go +++ b/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()) } diff --git a/client/rpc_client.go b/client/rpc_client.go index 724e3b89..a1f46dea 100644 --- a/client/rpc_client.go +++ b/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, diff --git a/server/buffer.go b/server/buffer.go deleted file mode 100644 index 4df03c27..00000000 --- a/server/buffer.go +++ /dev/null @@ -1,14 +0,0 @@ -package server - -import ( - "bytes" -) - -type buffer struct { - *bytes.Buffer -} - -func (b *buffer) Close() error { - b.Buffer.Reset() - return nil -} diff --git a/server/grpc/buffer.go b/server/grpc/buffer.go deleted file mode 100644 index c43bb231..00000000 --- a/server/grpc/buffer.go +++ /dev/null @@ -1,14 +0,0 @@ -package grpc - -import ( - "bytes" -) - -type buffer struct { - *bytes.Buffer -} - -func (b *buffer) Close() error { - b.Buffer.Reset() - return nil -} diff --git a/server/grpc/subscriber.go b/server/grpc/subscriber.go index 97b9d75e..3bd4abef 100644 --- a/server/grpc/subscriber.go +++ b/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() diff --git a/server/subscriber.go b/server/subscriber.go index 3d87fea1..e3f29286 100644 --- a/server/subscriber.go +++ b/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() diff --git a/transport/http_transport.go b/transport/http_transport.go index ec098ca2..b1e760a4 100644 --- a/transport/http_transport.go +++ b/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) - defer reqB.Reset() - buf := &buffer{ - reqB, - } + b := buf.New(bytes.NewBuffer(m.Body)) + defer b.Close() req := &http.Request{ Method: "POST", @@ -97,8 +87,8 @@ func (h *httpTransportClient) Send(m *Message) error { Host: h.addr, }, Header: header, - Body: buf, - ContentLength: int64(reqB.Len()), + Body: b, + ContentLength: int64(b.Len()), Host: h.addr, } diff --git a/client/buffer.go b/util/buf/buf.go similarity index 52% rename from client/buffer.go rename to util/buf/buf.go index 9eee2b3d..14f07cd2 100644 --- a/client/buffer.go +++ b/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} +}