Repository: goadesign/goa (go)

Issue:
The StreamingPayload generates broken code
My dsl code:

```
	Method("broker", func() {
		NoSecurity()
		Payload(func() {
			Field(1, "token", String, "JWTAuth token used to perform authorization", func() {
			})
			Field(2, "host_id", UInt64, "host_id", func() {
			})
			Required("host_id")
		})

		StreamingPayload(func() {
			Field(1, "type", String, "type value", func() {
				Enum("heartbeat", "keyboard")
			})
			Field(2, "keyboard_info", Keyboard, "keyboard info")
			Field(3, "ext", String, "other info")
		})
		StreamingResult(func() {
						Field(1, "type", String, "type value", func() {
				Enum("heartbeat", "keyboard")
			})
			Field(2, "keyboard_info", Keyboard, "keyboard info")
			Field(3, "ext", String, "other info")
		})
	
		Error("BadRequest")
		HTTP(func() {
			GET("/broker/{host_id}")
			Response(StatusOK)
			Header("token:Authorization", String, "Auth token", func() {
				//Pattern("^Bearer [^ ]+$")
			})
			Response("BadRequest", StatusBadRequest)
		})

		GRPC(func() {
			Response(CodeOK)		
			Response("BadRequest", CodeFailedPrecondition)
		})
	})
```

My code compile error info :

```
../../gen/http/keyboard/server/websocket.go:133:36: cannot use &body (type **BrokerStreamingBody) as type *BrokerStreamingBody in argument to ValidateBrokerStreamingBody


```

Pull request description:
Properly handle objects in streaming payloads
when there are validations on the type. The validation codegen expects
expects "body" to contain the actual object and not a pointer to it.
The streaming payload receive code uses pointers to check for nils
written by the client when using "RecvAndClose" and the body
variable contained a pointer making the call to the validation method
invalid. The changes make that case explicit by introducing a new
field in the data structure used to generate that code.

Fix #2664 

Patch:
diff --git a/http/codegen/websocket.go b/http/codegen/websocket.go
index 8b5c71fc..1c318da6 100644
--- a/http/codegen/websocket.go
+++ b/http/codegen/websocket.go
@@ -48,6 +48,11 @@ type (
 		// RecvTypeRef is the fully qualified type ref received from the
 		// stream.
 		RecvTypeRef string
+		// RecvTypeIsPointer is true if the type received from the stream is a
+		// array or map. This is needed so that the code reading the stream can
+		// use a pointer reference when needed to check whether anything was
+		// read (check against the nil value) and in this case return EOF.
+		RecvTypeIsPointer bool
 		// MustClose indicates whether to generate the Close() function
 		// for the stream.
 		MustClose bool
@@ -176,23 +181,24 @@ func initWebSocketData(ed *EndpointData, e *expr.HTTPEndpointExpr, sd *ServiceDa
 		}
 	}
 	ed.ServerWebSocket = &WebSocketData{
-		VarName:      md.ServerStream.VarName,
-		Interface:    fmt.Sprintf("%s.%s", svc.PkgName, md.ServerStream.Interface),
-		Endpoint:     ed,
-		Payload:      svrPayload,
-		Response:     ed.Result.Responses[0],
-		PkgName:      svc.PkgName,
-		Type:         "server",
-		Kind:         md.ServerStream.Kind,
-		SendName:     md.ServerStream.SendName,
-		SendDesc:     svrSendDesc,
-		SendTypeName: svrSendTypeName,
-		SendTypeRef:  svrSendTypeRef,
-		RecvName:     md.ServerStream.RecvName,
-		RecvDesc:     svrRecvDesc,
-		RecvTypeName: svrRecvTypeName,
-		RecvTypeRef:  svrRecvTypeRef,
-		MustClose:    md.ServerStream.MustClose,
+		VarName:           md.ServerStream.VarName,
+		Interface:         fmt.Sprintf("%s.%s", svc.PkgName, md.ServerStream.Interface),
+		Endpoint:          ed,
+		Payload:           svrPayload,
+		Response:          ed.Result.Responses[0],
+		PkgName:           svc.PkgName,
+		Type:              "server",
+		Kind:              md.ServerStream.Kind,
+		SendName:          md.ServerStream.SendName,
+		SendDesc:          svrSendDesc,
+		SendTypeName:      svrSendTypeName,
+		SendTypeRef:       svrSendTypeRef,
+		RecvName:          md.ServerStream.RecvName,
+		RecvDesc:          svrRecvDesc,
+		RecvTypeName:      svrRecvTypeName,
+		RecvTypeRef:       svrRecvTypeRef,
+		RecvTypeIsPointer: expr.IsArray(e.MethodExpr.StreamingPayload.Type) || expr.IsMap(e.MethodExpr.StreamingPayload.Type),
+		MustClose:         md.ServerStream.MustClose,
 	}
 	ed.ClientWebSocket = &WebSocketData{
 		VarName:      md.ClientStream.VarName,
@@ -570,7 +576,11 @@ func (s *{{ .VarName }}) {{ .RecvName }}() ({{ .RecvTypeRef }}, error) {
 	var (
 		rv {{ .RecvTypeRef }}
 	{{- if eq .Type "server" }}
-		msg *{{ .Payload.Ref }}
+		{{- if .RecvTypeIsPointer }}
+		body {{ .Payload.VarName }}
+		{{- else }}
+		msg *{{ .Payload.VarName }}
+		{{- end }}
 	{{- else }}
 		body {{ .Response.ClientBody.VarName }}
 	{{- end }}
@@ -578,23 +588,33 @@ func (s *{{ .VarName }}) {{ .RecvName }}() ({{ .RecvTypeRef }}, error) {
 	)
 {{- if eq .Type "server" }}
 	{{- template "websocket_upgrade" (upgradeParams .Endpoint .RecvName) }}
+	{{- if .RecvTypeIsPointer }}
+	if err = s.conn.ReadJSON(&body); err != nil {
+	{{- else }}
 	if err = s.conn.ReadJSON(&msg); err != nil {
+	{{- end }}
 		return rv, err
 	}
+	{{- if .RecvTypeIsPointer }}
+	if body == nil {
+	{{- else }}
 	if msg == nil {
+	{{- end }}
 		return rv, io.EOF
 	}
-	body := *msg
 	{{- if .Payload.ValidateRef }}
+		{{- if not .RecvTypeIsPointer }}
+	body := *msg
+		{{- end }}
 		{{ .Payload.ValidateRef }}
 		if err != nil {
 			return rv, err
 		}
 	{{- end }}
 	{{- if .Payload.Init }}
-		return {{ .Payload.Init.Name }}(body), nil
+		return {{ .Payload.Init.Name }}({{ if .RecvTypeIsPointer }}body{{ else }}msg{{ end }}), nil
 	{{- else }}
-		return body, nil
+		return {{ if .RecvTypeIsPointer }}body{{ else }}*msg{{ end }}, nil
 	{{- end }}
 {{- else }} {{/* client side code */}}
 	{{- if eq .RecvName "CloseAndRecv" }}


Test patch:
diff --git a/http/codegen/testdata/streaming_code.go b/http/codegen/testdata/streaming_code.go
index 28f55352..136e7343 100644
--- a/http/codegen/testdata/streaming_code.go
+++ b/http/codegen/testdata/streaming_code.go
@@ -967,7 +967,7 @@ var StreamingPayloadServerStreamRecvCode = `// Recv reads instances of "streamin
 func (s *StreamingPayloadMethodServerStream) Recv() (*streamingpayloadservice.Request, error) {
 	var (
 		rv  *streamingpayloadservice.Request
-		msg **StreamingPayloadMethodStreamingBody
+		msg *StreamingPayloadMethodStreamingBody
 		err error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
@@ -993,8 +993,7 @@ func (s *StreamingPayloadMethodServerStream) Recv() (*streamingpayloadservice.Re
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return NewStreamingPayloadMethodStreamingBody(body), nil
+	return NewStreamingPayloadMethodStreamingBody(msg), nil
 }
 `
 
@@ -1209,8 +1208,7 @@ func (s *StreamingPayloadNoResultMethodServerStream) Recv() (string, error) {
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1302,8 +1300,7 @@ func (s *StreamingPayloadResultWithViewsMethodServerStream) Recv() (float32, err
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1404,8 +1401,7 @@ func (s *StreamingPayloadResultWithExplicitViewMethodServerStream) Recv() (float
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1500,8 +1496,7 @@ func (s *StreamingPayloadResultCollectionWithViewsMethodServerStream) Recv() (in
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1607,8 +1602,7 @@ func (s *StreamingPayloadResultCollectionWithExplicitViewMethodServerStream) Rec
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1694,8 +1688,7 @@ func (s *StreamingPayloadPrimitiveMethodServerStream) Recv() (string, error) {
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -1746,9 +1739,9 @@ var StreamingPayloadPrimitiveArrayServerStreamRecvCode = `// Recv reads instance
 // "StreamingPayloadPrimitiveArrayMethod" endpoint websocket connection.
 func (s *StreamingPayloadPrimitiveArrayMethodServerStream) Recv() ([]int32, error) {
 	var (
-		rv  []int32
-		msg *[]int32
-		err error
+		rv   []int32
+		body []int32
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -1767,13 +1760,12 @@ func (s *StreamingPayloadPrimitiveArrayMethodServerStream) Recv() ([]int32, erro
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return body, nil
 }
 `
@@ -1825,9 +1817,9 @@ var StreamingPayloadPrimitiveMapServerStreamRecvCode = `// Recv reads instances
 // "StreamingPayloadPrimitiveMapMethod" endpoint websocket connection.
 func (s *StreamingPayloadPrimitiveMapMethodServerStream) Recv() (map[string]int32, error) {
 	var (
-		rv  map[string]int32
-		msg *map[string]int32
-		err error
+		rv   map[string]int32
+		body map[string]int32
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -1846,13 +1838,12 @@ func (s *StreamingPayloadPrimitiveMapMethodServerStream) Recv() (map[string]int3
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return body, nil
 }
 `
@@ -1905,9 +1896,9 @@ var StreamingPayloadUserTypeArrayServerStreamRecvCode = `// Recv reads instances
 // "StreamingPayloadUserTypeArrayMethod" endpoint websocket connection.
 func (s *StreamingPayloadUserTypeArrayMethodServerStream) Recv() ([]*streamingpayloadusertypearrayservice.RequestType, error) {
 	var (
-		rv  []*streamingpayloadusertypearrayservice.RequestType
-		msg *[]*RequestType
-		err error
+		rv   []*streamingpayloadusertypearrayservice.RequestType
+		body []*RequestType
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -1926,13 +1917,12 @@ func (s *StreamingPayloadUserTypeArrayMethodServerStream) Recv() ([]*streamingpa
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return NewStreamingPayloadUserTypeArrayMethodArray(body), nil
 }
 `
@@ -1987,9 +1977,9 @@ var StreamingPayloadUserTypeMapServerStreamRecvCode = `// Recv reads instances o
 // "StreamingPayloadUserTypeMapMethod" endpoint websocket connection.
 func (s *StreamingPayloadUserTypeMapMethodServerStream) Recv() (map[string]*streamingpayloadusertypemapservice.RequestType, error) {
 	var (
-		rv  map[string]*streamingpayloadusertypemapservice.RequestType
-		msg *map[string]*RequestType
-		err error
+		rv   map[string]*streamingpayloadusertypemapservice.RequestType
+		body map[string]*RequestType
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -2008,13 +1998,12 @@ func (s *StreamingPayloadUserTypeMapMethodServerStream) Recv() (map[string]*stre
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return NewStreamingPayloadUserTypeMapMethodMap(body), nil
 }
 `
@@ -2140,7 +2129,7 @@ var BidirectionalStreamingServerStreamRecvCode = `// Recv reads instances of "bi
 func (s *BidirectionalStreamingMethodServerStream) Recv() (*bidirectionalstreamingservice.Request, error) {
 	var (
 		rv  *bidirectionalstreamingservice.Request
-		msg **BidirectionalStreamingMethodStreamingBody
+		msg *BidirectionalStreamingMethodStreamingBody
 		err error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
@@ -2166,8 +2155,7 @@ func (s *BidirectionalStreamingMethodServerStream) Recv() (*bidirectionalstreami
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return NewBidirectionalStreamingMethodStreamingBody(body), nil
+	return NewBidirectionalStreamingMethodStreamingBody(msg), nil
 }
 `
 
@@ -2469,8 +2457,7 @@ func (s *BidirectionalStreamingResultWithViewsMethodServerStream) Recv() (float3
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -2613,8 +2600,7 @@ func (s *BidirectionalStreamingResultWithExplicitViewMethodServerStream) Recv()
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -2723,8 +2709,7 @@ func (s *BidirectionalStreamingResultCollectionWithViewsMethodServerStream) Recv
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -2841,8 +2826,7 @@ func (s *BidirectionalStreamingResultCollectionWithExplicitViewMethodServerStrea
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -2937,8 +2921,7 @@ func (s *BidirectionalStreamingPrimitiveMethodServerStream) Recv() (string, erro
 	if msg == nil {
 		return rv, io.EOF
 	}
-	body := *msg
-	return body, nil
+	return *msg, nil
 }
 `
 
@@ -2998,9 +2981,9 @@ var BidirectionalStreamingPrimitiveArrayServerStreamRecvCode = `// Recv reads in
 // "BidirectionalStreamingPrimitiveArrayMethod" endpoint websocket connection.
 func (s *BidirectionalStreamingPrimitiveArrayMethodServerStream) Recv() ([]int32, error) {
 	var (
-		rv  []int32
-		msg *[]int32
-		err error
+		rv   []int32
+		body []int32
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -3019,13 +3002,12 @@ func (s *BidirectionalStreamingPrimitiveArrayMethodServerStream) Recv() ([]int32
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return body, nil
 }
 `
@@ -3086,9 +3068,9 @@ var BidirectionalStreamingPrimitiveMapServerStreamRecvCode = `// Recv reads inst
 // "BidirectionalStreamingPrimitiveMapMethod" endpoint websocket connection.
 func (s *BidirectionalStreamingPrimitiveMapMethodServerStream) Recv() (map[string]int32, error) {
 	var (
-		rv  map[string]int32
-		msg *map[string]int32
-		err error
+		rv   map[string]int32
+		body map[string]int32
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -3107,13 +3089,12 @@ func (s *BidirectionalStreamingPrimitiveMapMethodServerStream) Recv() (map[strin
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return body, nil
 }
 `
@@ -3177,9 +3158,9 @@ var BidirectionalStreamingUserTypeArrayServerStreamRecvCode = `// Recv reads ins
 // "BidirectionalStreamingUserTypeArrayMethod" endpoint websocket connection.
 func (s *BidirectionalStreamingUserTypeArrayMethodServerStream) Recv() ([]*bidirectionalstreamingusertypearrayservice.RequestType, error) {
 	var (
-		rv  []*bidirectionalstreamingusertypearrayservice.RequestType
-		msg *[]*RequestType
-		err error
+		rv   []*bidirectionalstreamingusertypearrayservice.RequestType
+		body []*RequestType
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -3198,13 +3179,12 @@ func (s *BidirectionalStreamingUserTypeArrayMethodServerStream) Recv() ([]*bidir
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return NewBidirectionalStreamingUserTypeArrayMethodArray(body), nil
 }
 `
@@ -3272,9 +3252,9 @@ var BidirectionalStreamingUserTypeMapServerStreamRecvCode = `// Recv reads insta
 // "BidirectionalStreamingUserTypeMapMethod" endpoint websocket connection.
 func (s *BidirectionalStreamingUserTypeMapMethodServerStream) Recv() (map[string]*bidirectionalstreamingusertypemapservice.RequestType, error) {
 	var (
-		rv  map[string]*bidirectionalstreamingusertypemapservice.RequestType
-		msg *map[string]*RequestType
-		err error
+		rv   map[string]*bidirectionalstreamingusertypemapservice.RequestType
+		body map[string]*RequestType
+		err  error
 	)
 	// Upgrade the HTTP connection to a websocket connection only once. Connection
 	// upgrade is done here so that authorization logic in the endpoint is executed
@@ -3293,13 +3273,12 @@ func (s *BidirectionalStreamingUserTypeMapMethodServerStream) Recv() (map[string
 	if err != nil {
 		return rv, err
 	}
-	if err = s.conn.ReadJSON(&msg); err != nil {
+	if err = s.conn.ReadJSON(&body); err != nil {
 		return rv, err
 	}
-	if msg == nil {
+	if body == nil {
 		return rv, io.EOF
 	}
-	body := *msg
 	return NewBidirectionalStreamingUserTypeMapMethodMap(body), nil
 }
 `