unistack-org/protoc-gen-go-micro
4
0
Fork 0
Code Issues 5 Pull Requests Packages Projects Releases 61 Wiki Activity

Handle streams

Browse Source
This commit is contained in:
Manfred Touron 2016-12-23 19:20:44 +01:00
parent f4f2300417
commit 718f9120e4
No known key found for this signature in database
GPG Key ID: 9CCF47DF1FD978A1
3 changed files with 133 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

38
examples/go-kit/templates/{{.File.Package}}/gen/endpoints/endpoints.go.tmpl
View File

@ -12,15 +12,35 @@ import (
var _ = fmt.Errorf var _ = fmt.Errorf
type StreamEndpoint func(server interface{}, req interface{}) (err error)
type Endpoints struct { type Endpoints struct {
{{range .Service.Method}} {{range .Service.Method}}
{{if or (.ClientStreaming) (.ServerStreaming)}}
{{.Name}}Endpoint StreamEndpoint
{{else}}
{{.Name}}Endpoint endpoint.Endpoint {{.Name}}Endpoint endpoint.Endpoint
{{end}} {{end}}
{{end}}
} }
{{range .Service.Method}} {{range .Service.Method}}
/*{{. | prettyjson}}*/ {{if .ServerStreaming}}
{{if .ClientStreaming}}
func (e *Endpoints){{.Name}}(server pb.{{$file.Package | title}}Service_{{.Name}}Server) error {
return fmt.Errorf("not implemented")
}
{{else}}
func (e *Endpoints){{.Name}}(in *pb.{{.Name}}Request, server pb.{{$file.Package | title}}Service_{{.Name}}Server) error {
return fmt.Errorf("not implemented")
}
{{end}}
{{else}}
{{if .ClientStreaming}}
func (e *Endpoints){{.Name}}(server pb.{{$file.Package | title}}Service_{{.Name}}Server) error {
return fmt.Errorf("not implemented")
}
{{else}}
func (e *Endpoints){{.Name}}(ctx context.Context, in *pb.{{.InputType | splitArray "." | last}}) (*pb.{{.OutputType | splitArray "." | last}}, error) { func (e *Endpoints){{.Name}}(ctx context.Context, in *pb.{{.InputType | splitArray "." | last}}) (*pb.{{.OutputType | splitArray "." | last}}, error) {
out, err := e.{{.Name}}Endpoint(ctx, in) out, err := e.{{.Name}}Endpoint(ctx, in)
if err != nil { if err != nil {
@ -29,8 +49,21 @@ func (e *Endpoints){{.Name}}(ctx context.Context, in *pb.{{.InputType | splitArr
return out.(*pb.{{.OutputType | splitArray "." | last}}), err return out.(*pb.{{.OutputType | splitArray "." | last}}), err
} }
{{end}} {{end}}
{{end}}
{{end}}
{{range .Service.Method}} {{range .Service.Method}}
{{if or (.ServerStreaming) (.ClientStreaming)}}
func Make{{.Name}}Endpoint(svc pb.{{$file.Package | title}}ServiceServer) StreamEndpoint {
return func(server interface{}, request interface{}) error {
{{if .ClientStreaming}}
return svc.{{.Name}}(server.(pb.{{$file.Package | title}}Service_{{.Name}}Server))
{{else}}
return svc.{{.Name}}(request.(*pb.{{.Name}}Request), server.(pb.{{$file.Package | title}}Service_{{.Name}}Server))
{{end}}
}
}
{{else}}
func Make{{.Name}}Endpoint(svc pb.{{$file.Package | title}}ServiceServer) endpoint.Endpoint { func Make{{.Name}}Endpoint(svc pb.{{$file.Package | title}}ServiceServer) endpoint.Endpoint {
return func(ctx context.Context, request interface{}) (interface{}, error) { return func(ctx context.Context, request interface{}) (interface{}, error) {
req := request.(*pb.{{.InputType | splitArray "." | last}}) req := request.(*pb.{{.InputType | splitArray "." | last}})
@ -42,6 +75,7 @@ func Make{{.Name}}Endpoint(svc pb.{{$file.Package | title}}ServiceServer) endpoi
} }
} }
{{end}} {{end}}
{{end}}
func MakeEndpoints(svc pb.{{.File.Package | title}}ServiceServer) Endpoints { func MakeEndpoints(svc pb.{{.File.Package | title}}ServiceServer) Endpoints {
return Endpoints{ return Endpoints{

47
examples/go-kit/templates/{{.File.Package}}/gen/transports/grpc/grpc.go.tmpl
View File

@ -18,28 +18,43 @@ func MakeGRPCServer(ctx context.Context, endpoints endpoints.Endpoints) pb.{{.Fi
options := []grpctransport.ServerOption{} options := []grpctransport.ServerOption{}
return &grpcServer{ return &grpcServer{
{{range .Service.Method}} {{range .Service.Method}}
{{if not .ServerStreaming}} {{if or (.ClientStreaming) (.ServerStreaming)}}
{{if not .ClientStreaming}} {{.Name | lower}}: &server{
e: endpoints.{{.Name}}Endpoint,
},
{{else}}
{{.Name | lower}}: grpctransport.NewServer( {{.Name | lower}}: grpctransport.NewServer(
ctx, ctx,
endpoints.{{.Name}}Endpoint, endpoints.{{.Name}}Endpoint,
decode{{.Name}}Request, decodeRequest,
encode{{.Name}}Response, encode{{.Name}}Response,
options..., options...,
), ),
{{end}} {{end}}
{{end}} {{end}}
{{end}}
} }
} }
type grpcServer struct { type grpcServer struct {
{{range .Service.Method}} {{range .Service.Method}}
{{if or (.ClientStreaming) (.ServerStreaming)}}
{{.Name | lower}} streamHandler
{{else}}
{{.Name | lower}} grpctransport.Handler {{.Name | lower}} grpctransport.Handler
{{end}} {{end}}
{{end}}
} }
{{range .Service.Method}} {{range .Service.Method}}
{{if .ClientStreaming}}
func (s *grpcServer) {{.Name}}(server pb.{{$file.Package | title}}Service_{{.Name}}Server) error {
return s.{{.Name | lower}}.Do(server, nil)
}
{{else if .ServerStreaming}}
func (s *grpcServer) {{.Name}}(req *pb.{{.Name}}Request, server pb.{{$file.Package | title}}Service_{{.Name}}Server) error {
return s.{{.Name | lower}}.Do(server, req)
}
{{else}}
func (s *grpcServer) {{.Name}}(ctx context.Context, req *pb.{{.InputType | splitArray "." | last}}) (*pb.{{.OutputType | splitArray "." | last}}, error) { func (s *grpcServer) {{.Name}}(ctx context.Context, req *pb.{{.InputType | splitArray "." | last}}) (*pb.{{.OutputType | splitArray "." | last}}, error) {
_, rep, err := s.{{.Name | lower}}.ServeGRPC(ctx, req) _, rep, err := s.{{.Name | lower}}.ServeGRPC(ctx, req)
if err != nil { if err != nil {
@ -48,12 +63,28 @@ func (s *grpcServer) {{.Name}}(ctx context.Context, req *pb.{{.InputType | split
return rep.(*pb.{{.OutputType | splitArray "." | last}}), nil return rep.(*pb.{{.OutputType | splitArray "." | last}}), nil
} }
func decode{{.Name}}Request(ctx context.Context, grpcReq interface{}) (interface{}, error) {
return grpcReq, nil
}
func encode{{.Name}}Response(ctx context.Context, response interface{}) (interface{}, error) { func encode{{.Name}}Response(ctx context.Context, response interface{}) (interface{}, error) {
resp := response.(*pb.{{.OutputType | splitArray "." | last}}) resp := response.(*pb.{{.OutputType | splitArray "." | last}})
return resp, nil return resp, nil
} }
{{end}} {{end}}
{{end}}
func decodeRequest(ctx context.Context, grpcReq interface{}) (interface{}, error) {
return grpcReq, nil
}
type streamHandler interface{
Do(server interface{}, req interface{}) (err error)
}
type server struct {
e endpoints.StreamEndpoint
}
func (s server) Do(server interface{}, req interface{}) (err error) {
if err := s.e(server, req); err != nil {
return err
}
return nil
}

10
examples/go-kit/templates/{{.File.Package}}/gen/transports/http/http.go.tmpl
View File

@ -15,12 +15,13 @@ import (
) )
{{range .Service.Method}} {{range .Service.Method}}
{{if and (not .ServerStreaming) (not .ClientStreaming)}}
func Make{{.Name}}Handler(ctx context.Context, svc pb.{{$file.Package | title}}ServiceServer, endpoint gokit_endpoint.Endpoint) *httptransport.Server { func Make{{.Name}}Handler(ctx context.Context, svc pb.{{$file.Package | title}}ServiceServer, endpoint gokit_endpoint.Endpoint) *httptransport.Server {
return httptransport.NewServer( return httptransport.NewServer(
ctx, ctx,
endpoint, endpoint,
decode{{.Name}}Request, decode{{.Name}}Request,
encode{{.Name}}Response, encodeResponse,
[]httptransport.ServerOption{}..., []httptransport.ServerOption{}...,
) )
} }
@ -32,16 +33,19 @@ func decode{{.Name}}Request(ctx context.Context, r *http.Request) (interface{},
} }
return &req, nil return &req, nil
} }
{{end}}
{{end}}
func encode{{.Name}}Response(ctx context.Context, w http.ResponseWriter, response interface{}) error { func encodeResponse(ctx context.Context, w http.ResponseWriter, response interface{}) error {
return json.NewEncoder(w).Encode(response) return json.NewEncoder(w).Encode(response)
} }
{{end}}
func RegisterHandlers(ctx context.Context, svc pb.{{$file.Package | title}}ServiceServer, mux *http.ServeMux, endpoints endpoints.Endpoints) error { func RegisterHandlers(ctx context.Context, svc pb.{{$file.Package | title}}ServiceServer, mux *http.ServeMux, endpoints endpoints.Endpoints) error {
{{range .Service.Method}} {{range .Service.Method}}
{{if and (not .ServerStreaming) (not .ClientStreaming)}}
log.Println("new HTTP endpoint: \"/{{.Name}}\" (service={{$file.Package | title}})") log.Println("new HTTP endpoint: \"/{{.Name}}\" (service={{$file.Package | title}})")
mux.Handle("/{{.Name}}", Make{{.Name}}Handler(ctx, svc, endpoints.{{.Name}}Endpoint)) mux.Handle("/{{.Name}}", Make{{.Name}}Handler(ctx, svc, endpoints.{{.Name}}Endpoint))
{{end}} {{end}}
{{end}}
return nil return nil
} }