Merge pull request #102 from unistack-org/cleanup

server: remove unparsed body from request and message

.github/workflows/build.yml

@@ -12,7 +12,7 @@ jobs:
     - name: setup
       uses: actions/setup-go@v2
       with:
-        go-version: 1.16
+        go-version: 1.17
     - name: checkout
       uses: actions/checkout@v3
     - name: cache

.github/workflows/codeql-analysis.yml

@@ -47,7 +47,7 @@ jobs:
     - name: setup
       uses: actions/setup-go@v2
       with:
-        go-version: 1.16
+        go-version: 1.17
     # Initializes the CodeQL tools for scanning.
     - name: init
       uses: github/codeql-action/init@v1

.github/workflows/pr.yml

@@ -12,7 +12,7 @@ jobs:
     - name: setup
       uses: actions/setup-go@v2
       with:
-        go-version: 1.16
+        go-version: 1.17
     - name: checkout
       uses: actions/checkout@v3
     - name: cache

go.mod

@@ -4,7 +4,7 @@ go 1.16
 
 require (
 	github.com/ef-ds/deque v1.0.4
-	github.com/golang-jwt/jwt/v4 v4.3.0
+	github.com/golang-jwt/jwt/v4 v4.4.0
 	github.com/imdario/mergo v0.3.12
 	github.com/patrickmn/go-cache v2.1.0+incompatible
 	github.com/silas/dag v0.0.0-20211117232152-9d50aa809f35

go.sum

@@ -23,8 +23,8 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.m
 github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
 github.com/flowstack/go-jsonschema v0.1.1/go.mod h1:yL7fNggx1o8rm9RlgXv7hTBWxdBM0rVwpMwimd3F3N0=
 github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
-github.com/golang-jwt/jwt/v4 v4.3.0 h1:kHL1vqdqWNfATmA0FNMdmZNMyZI1U6O31X4rlIPoBog=
+github.com/golang-jwt/jwt/v4 v4.4.0 h1:EmVIxB5jzbllGIjiCV5JG4VylbK3KE400tLGLI1cdfU=
-github.com/golang-jwt/jwt/v4 v4.3.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
+github.com/golang-jwt/jwt/v4 v4.4.0/go.mod h1:/xlHOz8bRuivTWchD4jCa+NbatV+wEUSzwAxVc6locg=
 github.com/golang/glog v0.0.0-20160126235308-23def4e6c14b/go.mod h1:SBH7ygxi8pfUlaOkMMuAQtPIUF8ecWP5IEl/CR7VP2Q=
 github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
 github.com/golang/protobuf v1.2.0/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=

server/request.go

@@ -22,7 +22,7 @@ func (r *rpcMessage) Topic() string {
 	return r.topic
 }
 
-func (r *rpcMessage) Payload() interface{} {
+func (r *rpcMessage) Body() interface{} {
 	return r.payload
 }
 
@@ -30,10 +30,6 @@ func (r *rpcMessage) Header() metadata.Metadata {
 	return r.header
 }
 
-func (r *rpcMessage) Body() []byte {
-	return r.body
-}
-
 func (r *rpcMessage) Codec() codec.Codec {
 	return r.codec
 }

server/server.go

@@ -75,13 +75,11 @@ type Message interface {
 	// Topic of the message
 	Topic() string
 	// The decoded payload value
-	Payload() interface{}
+	Body() interface{}
 	// The content type of the payload
 	ContentType() string
 	// The raw headers of the message
 	Header() metadata.Metadata
-	// The raw body of the message
-	Body() []byte
 	// Codec used to decode the message
 	Codec() codec.Codec
 }

server/subscriber.go

@@ -252,7 +252,7 @@ func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.Bat
 					return err
 				}
 				rb := reflect.New(req.Type().Elem())
-				if err = cf.ReadBody(bytes.NewReader(msg.Body()), rb.Interface()); err != nil {
+				if err = cf.ReadBody(bytes.NewReader(msg.(*rpcMessage).body), rb.Interface()); err != nil {
 					return err
 				}
 				msg.(*rpcMessage).codec = cf
@@ -269,7 +269,7 @@ func (n *noopServer) newBatchSubHandler(sb *subscriber, opts Options) broker.Bat
 				}
 				payloads := reflect.MakeSlice(reqType, 0, len(ms))
 				for _, m := range ms {
-					payloads = reflect.Append(payloads, reflect.ValueOf(m.Payload()))
+					payloads = reflect.Append(payloads, reflect.ValueOf(m.Body()))
 				}
 				vals = append(vals, payloads)
 
@@ -381,7 +381,7 @@ func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler
 					vals = append(vals, reflect.ValueOf(ctx))
 				}
 
-				vals = append(vals, reflect.ValueOf(msg.Payload()))
+				vals = append(vals, reflect.ValueOf(msg.Body()))
 
 				returnValues := handler.method.Call(vals)
 				if rerr := returnValues[0].Interface(); rerr != nil {
@@ -406,7 +406,6 @@ func (n *noopServer) newSubHandler(sb *subscriber, opts Options) broker.Handler
 					contentType: ct,
 					payload:     req.Interface(),
 					header:      msg.Header,
-					body:        msg.Body,
 				})
 				results <- cerr
 			}()