func NewSingleHostProxy(endpoint string) *Proxy {
	return &Proxy{
		Endpoint: endpoint,
	}
}
NewSingleHostProxy returns a proxy which sends requests to a single backend
package mucp
import (
	"context"
	"fmt"
	"io"
	"sort"
	"strings"
	"sync"
	"time"
	"github.com/micro/micro/v3/service/client"
	grpcc "github.com/micro/micro/v3/service/client/grpc"
	"github.com/micro/micro/v3/service/context/metadata"
	"github.com/micro/micro/v3/service/errors"
	"github.com/micro/micro/v3/service/logger"
	"github.com/micro/micro/v3/service/proxy"
	"github.com/micro/micro/v3/service/router"
	"github.com/micro/micro/v3/service/router/registry"
	"github.com/micro/micro/v3/service/server"
	"github.com/micro/micro/v3/util/codec"
	"github.com/micro/micro/v3/util/codec/bytes"
	"github.com/micro/micro/v3/util/selector"
	"github.com/micro/micro/v3/util/selector/roundrobin"
	"google.golang.org/grpc"
)
type Proxy struct {
	options proxy.Options
	Endpoint string
	Client client.Client
	Links map[string]client.Client
	Router router.Router
	sync.RWMutex
	Routes map[string]map[uint64]router.Route
	Selector selector.Selector
}
func readLoop(r server.Request, s client.Stream) error {
	req := s.Request()
	for {
		body, err := r.Read()
		if err == io.EOF {
			return nil
		}
		if err != nil {
			return err
		}
		hdr := r.Header()
		msg := &codec.Message{
			Type:   codec.Request,
			Header: hdr,
			Body:   body,
		}
		err = req.Codec().Write(msg, nil)
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
	}
}
func toNodes(routes []router.Route) []string {
	nodes := make([]string, 0, len(routes))
	for _, node := range routes {
		address := node.Address
		if len(node.Gateway) > 0 {
			address = node.Gateway
		}
		nodes = append(nodes, address)
	}
	return nodes
}
func toSlice(r map[uint64]router.Route) []router.Route {
	routes := make([]router.Route, 0, len(r))
	for _, v := range r {
		routes = append(routes, v)
	}
	sort.Slice(routes, func(i, j int) bool { return routes[i].Metric < routes[j].Metric })
	return routes
}
func (p *Proxy) filterRoutes(ctx context.Context, routes []router.Route) []router.Route {
	md, ok := metadata.FromContext(ctx)
	if !ok {
		return routes
	}
	var filteredRoutes []router.Route
	for _, route := range routes {
		if id, ok := md.Get("Micro-Router"); ok && len(id) > 0 {
			if route.Router != id {
				continue
			}
		}
		if net, ok := md.Get("Micro-Namespace"); ok && len(net) > 0 {
			if route.Network != router.DefaultNetwork && route.Network != net {
				continue
			}
		}
		if gw, ok := md.Get("Micro-Gateway"); ok && len(gw) > 0 {
			if gw == p.Router.Options().Address {
				if len(route.Gateway) > 0 && route.Gateway != gw {
					continue
				}
			} else {
				if route.Gateway != gw {
					continue
				}
			}
		}
		filteredRoutes = append(filteredRoutes, route)
	}
	if logger.V(logger.TraceLevel, logger.DefaultLogger) {
		logger.Tracef("Proxy filtered routes %+v", filteredRoutes)
	}
	return filteredRoutes
}
func (p *Proxy) getLink(r router.Route) (client.Client, error) {
	if r.Link == "local" || len(p.Links) == 0 {
		return p.Client, nil
	}
	l, ok := p.Links[r.Link]
	if !ok {
		return nil, errors.InternalServerError("go.micro.proxy", "link not found")
	}
	return l, nil
}
func (p *Proxy) getRoute(ctx context.Context, service string) ([]router.Route, error) {
	p.RLock()
	cached, ok := p.Routes[service]
	p.RUnlock()
	if ok {
		return p.filterRoutes(ctx, toSlice(cached)), nil
	}
	routes, err := p.cacheRoutes(service)
	if err != nil {
		return nil, err
	}
	return p.filterRoutes(ctx, routes), nil
}
func (p *Proxy) cacheRoutes(service string) ([]router.Route, error) {
	results, err := p.Router.Lookup(service, router.LookupNetwork("*"))
	if err != nil {
		logger.Debugf("Failed to lookup route for %s: %v", service, err)
		return nil, err
	}
	p.Lock()
	delete(p.Routes, service)
	for _, route := range results {
		if _, ok := p.Routes[service]; !ok {
			p.Routes[service] = make(map[uint64]router.Route)
		}
		p.Routes[service][route.Hash()] = route
	}
	routes := p.Routes[service]
	p.Unlock()
	return toSlice(routes), nil
}
func (p *Proxy) refreshMetrics() {
	p.RLock()
	services := make([]string, 0, len(p.Routes))
	for service := range p.Routes {
		services = append(services, service)
	}
	p.RUnlock()
	for _, service := range services {
		p.cacheRoutes(service)
	}
}
func (p *Proxy) manageRoutes(route router.Route, action string) error {
	p.Lock()
	defer p.Unlock()
	if logger.V(logger.TraceLevel, logger.DefaultLogger) {
		logger.Tracef("Proxy taking route action %v %+v\n", action, route)
	}
	switch action {
	case "create", "update":
		if _, ok := p.Routes[route.Service]; !ok {
			return fmt.Errorf("not called %s", route.Service)
		}
		p.Routes[route.Service][route.Hash()] = route
	case "delete":
		delete(p.Routes[route.Service], route.Hash())
		if len(p.Routes[route.Service]) == 0 {
			delete(p.Routes, route.Service)
		}
	default:
		return fmt.Errorf("unknown action: %s", action)
	}
	return nil
}
func (p *Proxy) watchRoutes() {
	w, err := p.Router.Watch()
	if err != nil {
		logger.Debugf("Error watching router: %v", err)
		return
	} else if w == nil {
		logger.Debugf("Error watching routes, no watcher returned")
		return
	}
	defer w.Stop()
	for {
		event, err := w.Next()
		if err != nil {
			logger.Debugf("Error watching router: %v", err)
			return
		}
		if err := p.manageRoutes(event.Route, event.Type.String()); err != nil {
			continue
		}
	}
}
func (p *Proxy) ProcessMessage(ctx context.Context, msg server.Message) error {
	if logger.V(logger.TraceLevel, logger.DefaultLogger) {
		logger.Tracef("Proxy received message for %s", msg.Topic())
	}
	var errors []string
	if err := p.Client.Publish(ctx, msg); err != nil {
		errors = append(errors, err.Error())
	}
	for _, client := range p.Links {
		if err := client.Publish(ctx, msg); err != nil {
			errors = append(errors, err.Error())
		}
	}
	if len(errors) == 0 {
		return nil
	}
	return fmt.Errorf("Message processing error: %s", strings.Join(errors, "\n"))
}
func (p *Proxy) ServeRequest(ctx context.Context, req server.Request, rsp server.Response) error {
	var local bool
	var addresses []string
	var routes []router.Route
	service := req.Service()
	endpoint := req.Endpoint()
	if len(service) == 0 {
		return errors.BadRequest("go.micro.proxy", "service name is blank")
	}
	if logger.V(logger.TraceLevel, logger.DefaultLogger) {
		logger.Tracef("Proxy received request for %s %s", service, endpoint)
	}
	if len(p.Links) == 0 {
		local = true
	}
	if len(p.Client.Options().Proxy) > 0 {
		return p.serveRequest(ctx, p.Client, service, endpoint, req, rsp)
	}
	if len(p.Endpoint) > 0 {
		if parts := strings.Split(p.Endpoint, ":"); len(parts) > 1 {
			addresses = []string{p.Endpoint}
		} else {
			addr, err := p.getRoute(ctx, p.Endpoint)
			if err != nil {
				return err
			}
			routes = addr
			service = p.Endpoint
		}
	} else {
		addr, err := p.getRoute(ctx, service)
		if err != nil {
			return err
		}
		routes = addr
	}
	opts := []client.CallOption{
		client.WithSelector(p.Selector),
	}
	if len(addresses) > 0 {
		opts = append(opts,
			client.WithAddress(addresses...),
		)
		return p.serveRequest(ctx, p.Client, service, endpoint, req, rsp, opts...)
	}
	if local {
		var opts []client.CallOption
		if len(routes) > 0 {
			addresses = toNodes(routes)
			opts = append(opts, client.WithAddress(addresses...))
		}
		if logger.V(logger.TraceLevel, logger.DefaultLogger) {
			logger.Tracef("Proxy calling %+v\n", addresses)
		}
		return p.serveRequest(ctx, p.Client, service, endpoint, req, rsp, opts...)
	}
	if len(routes) == 0 {
		return errors.InternalServerError("go.micro.proxy", "route not found")
	}
	var gerr error
	for _, route := range routes {
		link, err := p.getLink(route)
		if err != nil {
			gerr = err
			continue
		}
		if logger.V(logger.TraceLevel, logger.DefaultLogger) {
			logger.Tracef("Proxy using route %+v\n", route)
		}
		addresses := toNodes([]router.Route{route})
		opts = append(opts,
			client.WithAddress(addresses...),
			client.WithRetries(0),
		)
		gerr = p.serveRequest(ctx, link, service, endpoint, req, rsp, opts...)
		if gerr == nil {
			return nil
		}
		if gerr == context.Canceled || gerr == context.DeadlineExceeded {
			return err
		}
	}
	return gerr
}
func (p *Proxy) serveRequest(ctx context.Context, link client.Client, service, endpoint string, req server.Request, rsp server.Response, opts ...client.CallOption) error {
	body, err := req.Read()
	if err != nil {
		return err
	}
	creq := link.NewRequest(service, endpoint, &bytes.Frame{Data: body}, client.WithContentType(req.ContentType()))
	if !req.Stream() {
		crsp := new(bytes.Frame)
		if err := link.Call(ctx, creq, crsp, opts...); err != nil {
			return err
		}
		if err := rsp.Write(crsp.Data); err != nil {
			return err
		}
		return nil
	}
	ctx, cancel := context.WithCancel(ctx)
	stream, err := link.Stream(ctx, creq, opts...)
	if err != nil {
		return err
	}
	defer stream.Close()
	c, ok := req.Codec().(codec.Codec)
	if ok && c.String() == "grpc" && link.String() == "grpc" {
		hdr := req.Header()
		msg := &codec.Message{
			Type:   codec.Request,
			Header: hdr,
			Body:   body,
		}
		err = stream.Request().Codec().Write(msg, nil)
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
	}
	go func() {
		err := readLoop(req, stream)
		if err != nil && err != io.EOF {
			cancel()
		}
	}()
	resp := stream.Response()
	for {
		body, err := resp.Read()
		if err != nil {
			if cc, ok := stream.(grpc.ClientStream); ok {
				if ss, ok := resp.Codec().(grpc.ServerStream); ok {
					ss.SetTrailer(cc.Trailer())
				}
			}
		}
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
		hdr := resp.Header()
		rsp.WriteHeader(hdr)
		err = rsp.Write(body)
		if err == io.EOF {
			return nil
		} else if err != nil {
			return err
		}
	}
}
func (p *Proxy) String() string {
	return "mucp"
}