Merge pull request grpc-ecosystem#170 from tmc/concurrent_forward

Add bidirectional streaming support by running Send() and Recv() concurrently
adasari · Jun 13, 2016 · 687fc9d · 687fc9d
2 parents d4b55f1 + c8dd1b7
commit 687fc9d
Show file tree

Hide file tree

Showing 11 changed files with 103 additions and 23 deletions.
diff --git a/examples/examplepb/a_bit_of_everything.pb.go b/examples/examplepb/a_bit_of_everything.pb.go
diff --git a/examples/examplepb/a_bit_of_everything.pb.gw.go b/examples/examplepb/a_bit_of_everything.pb.gw.go
diff --git a/examples/examplepb/echo_service.pb.go b/examples/examplepb/echo_service.pb.go
diff --git a/examples/examplepb/flow_combination.pb.go b/examples/examplepb/flow_combination.pb.go
diff --git a/examples/examplepb/flow_combination.pb.gw.go b/examples/examplepb/flow_combination.pb.gw.go
diff --git a/examples/sub/message.pb.go b/examples/sub/message.pb.go
diff --git a/examples/sub2/message.pb.go b/examples/sub2/message.pb.go
diff --git a/protoc-gen-grpc-gateway/gengateway/template.go b/protoc-gen-grpc-gateway/gengateway/template.go
@@ -117,7 +117,9 @@ var _ = utilities.NewDoubleArray
 `))
 
 	handlerTemplate = template.Must(template.New("handler").Parse(`
-{{if .Method.GetClientStreaming}}
+{{if and .Method.GetClientStreaming .Method.GetServerStreaming}}
+{{template "bidi-streaming-request-func" .}}
+{{else if .Method.GetClientStreaming}}
 {{template "client-streaming-request-func" .}}
 {{else}}
 {{template "client-rpc-request-func" .}}
@@ -234,6 +236,54 @@ var (
 {{end}}
 }`))
 
+	_ = template.Must(handlerTemplate.New("bidi-streaming-request-func").Parse(`
+{{template "request-func-signature" .}} {
+	var metadata runtime.ServerMetadata
+	stream, err := client.{{.Method.GetName}}(ctx)
+	if err != nil {
+		grpclog.Printf("Failed to start streaming: %v", err)
+		return nil, metadata, err
+	}
+	dec := marshaler.NewDecoder(req.Body)
+	handleSend := func() error {
+		var protoReq {{.Method.RequestType.GoType .Method.Service.File.GoPkg.Path}}
+		err = dec.Decode(&protoReq)
+		if err != nil {
+			grpclog.Printf("Failed to decode request: %v", err)
+			return err
+		}
+		if err = stream.Send(&protoReq); err != nil {
+			grpclog.Printf("Failed to send request: %v", err)
+			return err
+		}
+		return nil
+	}
+	if err := handleSend(); err != nil {
+		if err := stream.CloseSend(); err != nil {
+			grpclog.Printf("Failed to terminate client stream: %v", err)
+		}
+		return nil, metadata, err
+	}
+	go func() {
+		for {
+			if err := handleSend(); err != nil {
+				break
+			}
+		}
+		if err := stream.CloseSend(); err != nil {
+			grpclog.Printf("Failed to terminate client stream: %v", err)
+		}
+	}()
+	header, err := stream.Header()
+	if err != nil {
+		grpclog.Printf("Failed to get header from client: %v", err)
+		return nil, metadata, err
+	}
+	metadata.HeaderMD = header
+	return stream, metadata, nil
+}
+`))
+
 	trailerTemplate = template.Must(template.New("trailer").Parse(`
 {{range $svc := .}}
 // Register{{$svc.GetName}}HandlerFromEndpoint is same as Register{{$svc.GetName}}Handler but

diff --git a/runtime/stream_chunk.pb.go b/runtime/stream_chunk.pb.go
diff --git a/third_party/googleapis/google/api/annotations.pb.go b/third_party/googleapis/google/api/annotations.pb.go
diff --git a/third_party/googleapis/google/api/http.pb.go b/third_party/googleapis/google/api/http.pb.go