Merge pull request #30 from moul/dev/moul/handle-streams
Go-kit example: Handle streams
This commit is contained in:
commit
4c53f8ebdd
@ -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{
|
||||||
|
@ -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
|
||||||
|
}
|
||||||
|
@ -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
|
||||||
}
|
}
|
||||||
|
Loading…
Reference in New Issue
Block a user