rename

server/rpc_codec.go

@@ -29,6 +29,13 @@ type readWriteCloser struct {
 	rbuf *bytes.Buffer
 }
 
+type serverCodec interface {
+	ReadHeader(*request, bool) error
+	ReadBody(interface{}) error
+	Write(*response, interface{}, bool) error
+	Close() error
+}
+
 var (
 	DefaultContentType = "application/protobuf"
 
@@ -72,7 +79,7 @@ func newRpcCodec(req *transport.Message, socket transport.Socket, c codec.NewCod
 	return r
 }
 
-func (c *rpcCodec) ReadRequestHeader(r *request, first bool) error {
+func (c *rpcCodec) ReadHeader(r *request, first bool) error {
 	m := codec.Message{Header: c.req.Header}
 
 	if !first {
@@ -106,11 +113,11 @@ func (c *rpcCodec) ReadRequestHeader(r *request, first bool) error {
 	return err
 }
 
-func (c *rpcCodec) ReadRequestBody(b interface{}) error {
+func (c *rpcCodec) ReadBody(b interface{}) error {
 	return c.codec.ReadBody(b)
 }
 
-func (c *rpcCodec) WriteResponse(r *response, body interface{}, last bool) error {
+func (c *rpcCodec) Write(r *response, body interface{}, last bool) error {
 	c.buf.wbuf.Reset()
 	m := &codec.Message{
 		Method: r.ServiceMethod,

server/rpc_codec_test.go

@@ -47,7 +47,7 @@ func TestCodecWriteError(t *testing.T) {
 		socket: socket,
 	}
 
-	err := c.WriteResponse(&response{
+	err := c.Write(&response{
 		ServiceMethod: "Service.Method",
 		Seq:           0,
 		Error:         "",
@@ -55,7 +55,7 @@ func TestCodecWriteError(t *testing.T) {
 	}, "body", false)
 
 	if err != nil {
-		t.Fatalf(`Expected WriteResponse to fail; got "%+v" instead`, err)
+		t.Fatalf(`Expected Write to fail; got "%+v" instead`, err)
 	}
 
 	const expectedError = "Unable to encode body: simulating a codec write failure"

server/rpc_router.go

@@ -225,7 +225,7 @@ func (router *router) sendResponse(sending sync.Locker, req *request, reply inte
 	}
 	resp.Seq = req.Seq
 	sending.Lock()
-	err = codec.WriteResponse(resp, reply, last)
+	err = codec.Write(resp, reply, last)
 	sending.Unlock()
 	router.freeResponse(resp)
 	return err
@@ -391,12 +391,12 @@ func (router *router) readRequest(codec serverCodec) (service *service, mtype *m
 			return
 		}
 		// discard body
-		codec.ReadRequestBody(nil)
+		codec.ReadBody(nil)
 		return
 	}
 	// is it a streaming request? then we don't read the body
 	if mtype.stream {
-		codec.ReadRequestBody(nil)
+		codec.ReadBody(nil)
 		return
 	}
 
@@ -409,7 +409,7 @@ func (router *router) readRequest(codec serverCodec) (service *service, mtype *m
 		argIsValue = true
 	}
 	// argv guaranteed to be a pointer now.
-	if err = codec.ReadRequestBody(argv.Interface()); err != nil {
+	if err = codec.ReadBody(argv.Interface()); err != nil {
 		return
 	}
 	if argIsValue {
@@ -425,7 +425,7 @@ func (router *router) readRequest(codec serverCodec) (service *service, mtype *m
 func (router *router) readRequestHeader(codec serverCodec) (service *service, mtype *methodType, req *request, keepReading bool, err error) {
 	// Grab the request header.
 	req = router.getRequest()
-	err = codec.ReadRequestHeader(req, true)
+	err = codec.ReadHeader(req, true)
 	if err != nil {
 		req = nil
 		if err == io.EOF || err == io.ErrUnexpectedEOF {
@@ -458,11 +458,3 @@ func (router *router) readRequestHeader(codec serverCodec) (service *service, mt
 	}
 	return
 }
-
-type serverCodec interface {
-	ReadRequestHeader(*request, bool) error
-	ReadRequestBody(interface{}) error
-	WriteResponse(*response, interface{}, bool) error
-
-	Close() error
-}

server/rpc_server.go

@@ -70,6 +70,26 @@ func (s *rpcServer) accept(sock transport.Socket) {
 		// we use this Content-Type header to identify the codec needed
 		ct := msg.Header["Content-Type"]
 
+		// strip our headers
+		hdr := make(map[string]string)
+		for k, v := range msg.Header {
+			hdr[k] = v
+		}
+
+		// set local/remote ips
+		hdr["Local"] = sock.Local()
+		hdr["Remote"] = sock.Remote()
+
+		// create new context
+		ctx := metadata.NewContext(context.Background(), hdr)
+
+		// set the timeout if we have it
+		if len(to) > 0 {
+			if n, err := strconv.ParseUint(to, 10, 64); err == nil {
+				ctx, _ = context.WithTimeout(ctx, time.Duration(n))
+			}
+		}
+
 		// no content type
 		if len(ct) == 0 {
 			ct = DefaultContentType
@@ -88,29 +108,9 @@ func (s *rpcServer) accept(sock transport.Socket) {
 			return
 		}
 
+		// create the internal server codec
 		codec := newRpcCodec(&msg, sock, cf)
 
-		// strip our headers
-		hdr := make(map[string]string)
-		for k, v := range msg.Header {
-			hdr[k] = v
-		}
-		delete(hdr, "Content-Type")
-		delete(hdr, "Timeout")
-
-		// set local/remote ips
-		hdr["Local"] = sock.Local()
-		hdr["Remote"] = sock.Remote()
-
-		ctx := metadata.NewContext(context.Background(), hdr)
-
-		// set the timeout if we have it
-		if len(to) > 0 {
-			if n, err := strconv.ParseUint(to, 10, 64); err == nil {
-				ctx, _ = context.WithTimeout(ctx, time.Duration(n))
-			}
-		}
-
 		// TODO: needs better error handling
 		if err := s.router.ServeRequest(ctx, codec, ct); err != nil {
 			s.wg.Done()

server/rpc_stream.go

@@ -33,7 +33,7 @@ func (r *rpcStream) Send(msg interface{}) error {
 		Seq:           r.seq,
 	}
 
-	return r.codec.WriteResponse(&resp, msg, false)
+	return r.codec.Write(&resp, msg, false)
 }
 
 func (r *rpcStream) Recv(msg interface{}) error {
@@ -42,15 +42,15 @@ func (r *rpcStream) Recv(msg interface{}) error {
 
 	req := request{}
 
-	if err := r.codec.ReadRequestHeader(&req, false); err != nil {
+	if err := r.codec.ReadHeader(&req, false); err != nil {
 		// discard body
-		r.codec.ReadRequestBody(nil)
+		r.codec.ReadBody(nil)
 		return err
 	}
 
 	// we need to stay up to date with sequence numbers
 	r.seq = req.Seq
-	return r.codec.ReadRequestBody(msg)
+	return r.codec.ReadBody(msg)
 }
 
 func (r *rpcStream) Error() error {

server/server.go

@@ -9,6 +9,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/micro/go-log"
+	"github.com/micro/go-micro/codec"
 	"github.com/micro/go-micro/registry"
 )
 
@@ -29,7 +30,7 @@ type Server interface {
 
 // Router handle serving messages
 type Router interface {
-	ServeRequest(context.Context, Stream) error
+	ServeCodec(context.Context, codec.Codec) error
 }
 
 // Message is an async message interface