add proxy interface and move router packages
This commit is contained in:
246
proxy/grpc/grpc.go
Normal file
246
proxy/grpc/grpc.go
Normal file
@@ -0,0 +1,246 @@
|
||||
// Package grpc transparently forwards the grpc protocol using a go-micro client.
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"strings"
|
||||
|
||||
"github.com/micro/go-micro"
|
||||
"github.com/micro/go-micro/client"
|
||||
"github.com/micro/go-micro/codec"
|
||||
"github.com/micro/go-micro/codec/bytes"
|
||||
"github.com/micro/go-micro/init"
|
||||
"github.com/micro/go-micro/server"
|
||||
"github.com/micro/go-micro/service/grpc"
|
||||
)
|
||||
|
||||
// Proxy will transparently proxy requests to the backend.
|
||||
// If no backend is specified it will call a service using the client.
|
||||
// If the service matches the Name it will use the server.DefaultProxy.
|
||||
type Proxy struct {
|
||||
// Name of the local service. In the event it's to be left alone
|
||||
Name string
|
||||
|
||||
// Backend is a single backend to route to
|
||||
// If backend is of the form address:port it will call the address.
|
||||
// Otherwise it will use it as the service name to call.
|
||||
Backend string
|
||||
|
||||
// Endpoint specified the fixed endpoint to call.
|
||||
// In the event you proxy to a fixed backend this lets you
|
||||
// call a single endpoint
|
||||
Endpoint string
|
||||
|
||||
// The client to use for outbound requests
|
||||
Client client.Client
|
||||
|
||||
// The proxy options
|
||||
Options init.Options
|
||||
}
|
||||
|
||||
var (
|
||||
// The default name of this local service
|
||||
DefaultName = "go.micro.proxy"
|
||||
// The default router
|
||||
DefaultProxy = &Proxy{}
|
||||
)
|
||||
|
||||
// read client request and write to server
|
||||
func readLoop(r server.Request, s client.Stream) error {
|
||||
// request to backend server
|
||||
req := s.Request()
|
||||
|
||||
for {
|
||||
// get data from client
|
||||
// no need to decode it
|
||||
body, err := r.Read()
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
}
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// get the header from client
|
||||
hdr := r.Header()
|
||||
msg := &codec.Message{
|
||||
Type: codec.Request,
|
||||
Header: hdr,
|
||||
Body: body,
|
||||
}
|
||||
// write the raw request
|
||||
err = req.Codec().Write(msg, nil)
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ServeRequest honours the server.Proxy interface
|
||||
func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server.Response) error {
|
||||
// set the default name e.g local proxy
|
||||
if p.Name == "" {
|
||||
p.Name = DefaultName
|
||||
}
|
||||
|
||||
// set default client
|
||||
if p.Client == nil {
|
||||
p.Client = client.DefaultClient
|
||||
}
|
||||
|
||||
// check service route
|
||||
if req.Service() == p.Name {
|
||||
// use the default router
|
||||
return server.DefaultProxy.ServeRequest(ctx, req, rsp)
|
||||
}
|
||||
|
||||
opts := []client.CallOption{}
|
||||
|
||||
// service name
|
||||
service := req.Service()
|
||||
endpoint := req.Endpoint()
|
||||
|
||||
// call a specific backend
|
||||
if len(p.Backend) > 0 {
|
||||
// address:port
|
||||
if parts := strings.Split(p.Backend, ":"); len(parts) > 0 {
|
||||
opts = append(opts, client.WithAddress(p.Backend))
|
||||
// use as service name
|
||||
} else {
|
||||
service = p.Backend
|
||||
}
|
||||
}
|
||||
|
||||
// call a specific endpoint
|
||||
if len(p.Endpoint) > 0 {
|
||||
endpoint = p.Endpoint
|
||||
}
|
||||
|
||||
// read initial request
|
||||
body, err := req.Read()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// create new request with raw bytes body
|
||||
creq := p.Client.NewRequest(service, endpoint, &bytes.Frame{body}, client.WithContentType(req.ContentType()))
|
||||
|
||||
// create new stream
|
||||
stream, err := p.Client.Stream(ctx, creq, opts...)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stream.Close()
|
||||
|
||||
// create client request read loop
|
||||
go readLoop(req, stream)
|
||||
|
||||
// get raw response
|
||||
resp := stream.Response()
|
||||
|
||||
// create server response write loop
|
||||
for {
|
||||
// read backend response body
|
||||
body, err := resp.Read()
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// read backend response header
|
||||
hdr := resp.Header()
|
||||
|
||||
// write raw response header to client
|
||||
rsp.WriteHeader(hdr)
|
||||
|
||||
// write raw response body to client
|
||||
err = rsp.Write(body)
|
||||
if err == io.EOF {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// NewProxy returns a new grpc proxy server
|
||||
func NewProxy(opts ...init.Option) *Proxy {
|
||||
return &Proxy{
|
||||
Options: init.NewOptions(opts...),
|
||||
}
|
||||
}
|
||||
|
||||
// NewSingleHostProxy returns a router which sends requests to a single backend
|
||||
//
|
||||
// It is used by setting it in a new micro service to act as a proxy for a backend.
|
||||
//
|
||||
// Usage:
|
||||
//
|
||||
// Create a new router to the http backend
|
||||
//
|
||||
// r := NewSingleHostProxy("localhost:10001")
|
||||
//
|
||||
// // Create your new service
|
||||
// service := micro.NewService(
|
||||
// micro.Name("greeter"),
|
||||
// // Set the router
|
||||
// http.WithProxy(r),
|
||||
// )
|
||||
//
|
||||
// // Run the service
|
||||
// service.Run()
|
||||
func NewSingleHostProxy(url string) *Proxy {
|
||||
return &Proxy{
|
||||
Backend: url,
|
||||
}
|
||||
}
|
||||
|
||||
// NewService returns a new proxy. It acts as a micro service proxy.
|
||||
// Any request on the transport is routed to via the client to a service.
|
||||
// In the event a backend is specified then it routes to that backend.
|
||||
// The name of the backend can be a local address:port or a service name.
|
||||
//
|
||||
// Usage:
|
||||
//
|
||||
// New micro proxy routes via micro client to any service
|
||||
//
|
||||
// proxy := NewService()
|
||||
//
|
||||
// OR with address:port routes to local service
|
||||
//
|
||||
// service := NewService(
|
||||
// // Sets the default http endpoint
|
||||
// proxy.WithBackend("localhost:10001"),
|
||||
// )
|
||||
//
|
||||
// OR with service name routes to a fixed backend service
|
||||
//
|
||||
// service := NewService(
|
||||
// // Sets the backend service
|
||||
// proxy.WithBackend("greeter"),
|
||||
// )
|
||||
//
|
||||
func NewService(opts ...micro.Option) micro.Service {
|
||||
router := DefaultProxy
|
||||
name := DefaultName
|
||||
|
||||
// prepend router to opts
|
||||
opts = append([]micro.Option{
|
||||
micro.Name(name),
|
||||
WithProxy(router),
|
||||
}, opts...)
|
||||
|
||||
// create the new service
|
||||
service := grpc.NewService(opts...)
|
||||
|
||||
// set router name
|
||||
router.Name = service.Server().Options().Name
|
||||
|
||||
return service
|
||||
}
|
32
proxy/grpc/options.go
Normal file
32
proxy/grpc/options.go
Normal file
@@ -0,0 +1,32 @@
|
||||
package grpc
|
||||
|
||||
import (
|
||||
"github.com/micro/go-micro"
|
||||
"github.com/micro/go-micro/server"
|
||||
)
|
||||
|
||||
// WithBackend provides an option to set the proxy backend url
|
||||
func WithBackend(url string) micro.Option {
|
||||
return func(o *micro.Options) {
|
||||
// get the router
|
||||
r := o.Server.Options().Router
|
||||
|
||||
// not set
|
||||
if r == nil {
|
||||
r = DefaultRouter
|
||||
o.Server.Init(server.WithRouter(r))
|
||||
}
|
||||
|
||||
// check its a proxy router
|
||||
if proxyRouter, ok := r.(*Router); ok {
|
||||
proxyRouter.Backend = url
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// WithRouter provides an option to set the proxy router
|
||||
func WithRouter(r server.Router) micro.Option {
|
||||
return func(o *micro.Options) {
|
||||
o.Server.Init(server.WithRouter(r))
|
||||
}
|
||||
}
|
Reference in New Issue
Block a user