123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101 |
- // +build !confonly
- package command
- //go:generate go run github.com/xtls/xray-core/common/errors/errorgen
- import (
- "context"
- "time"
- "google.golang.org/grpc"
- "github.com/xtls/xray-core/common"
- "github.com/xtls/xray-core/core"
- "github.com/xtls/xray-core/features/routing"
- "github.com/xtls/xray-core/features/stats"
- )
- // routingServer is an implementation of RoutingService.
- type routingServer struct {
- router routing.Router
- routingStats stats.Channel
- }
- // NewRoutingServer creates a statistics service with statistics manager.
- func NewRoutingServer(router routing.Router, routingStats stats.Channel) RoutingServiceServer {
- return &routingServer{
- router: router,
- routingStats: routingStats,
- }
- }
- func (s *routingServer) TestRoute(ctx context.Context, request *TestRouteRequest) (*RoutingContext, error) {
- if request.RoutingContext == nil {
- return nil, newError("Invalid routing request.")
- }
- route, err := s.router.PickRoute(AsRoutingContext(request.RoutingContext))
- if err != nil {
- return nil, err
- }
- if request.PublishResult && s.routingStats != nil {
- ctx, _ := context.WithTimeout(context.Background(), 4*time.Second)
- s.routingStats.Publish(ctx, route)
- }
- return AsProtobufMessage(request.FieldSelectors)(route), nil
- }
- func (s *routingServer) SubscribeRoutingStats(request *SubscribeRoutingStatsRequest, stream RoutingService_SubscribeRoutingStatsServer) error {
- if s.routingStats == nil {
- return newError("Routing statistics not enabled.")
- }
- genMessage := AsProtobufMessage(request.FieldSelectors)
- subscriber, err := stats.SubscribeRunnableChannel(s.routingStats)
- if err != nil {
- return err
- }
- defer stats.UnsubscribeClosableChannel(s.routingStats, subscriber)
- for {
- select {
- case value, ok := <-subscriber:
- if !ok {
- return newError("Upstream closed the subscriber channel.")
- }
- route, ok := value.(routing.Route)
- if !ok {
- return newError("Upstream sent malformed statistics.")
- }
- err := stream.Send(genMessage(route))
- if err != nil {
- return err
- }
- case <-stream.Context().Done():
- return stream.Context().Err()
- }
- }
- }
- func (s *routingServer) mustEmbedUnimplementedRoutingServiceServer() {}
- type service struct {
- v *core.Instance
- }
- func (s *service) Register(server *grpc.Server) {
- common.Must(s.v.RequireFeatures(func(router routing.Router, stats stats.Manager) {
- rs := NewRoutingServer(router, nil)
- RegisterRoutingServiceServer(server, rs)
- // For compatibility purposes
- vCoreDesc := _RoutingService_serviceDesc
- vCoreDesc.ServiceName = "v2ray.core.app.router.command.RoutingService"
- server.RegisterService(&vCoreDesc, rs)
- }))
- }
- func init() {
- common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, cfg interface{}) (interface{}, error) {
- s := core.MustFromContext(ctx)
- return &service{v: s}, nil
- }))
- }
|