command.go 2.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101
  1. // +build !confonly
  2. package command
  3. //go:generate go run github.com/xtls/xray-core/common/errors/errorgen
  4. import (
  5. "context"
  6. "time"
  7. "google.golang.org/grpc"
  8. "github.com/xtls/xray-core/common"
  9. "github.com/xtls/xray-core/core"
  10. "github.com/xtls/xray-core/features/routing"
  11. "github.com/xtls/xray-core/features/stats"
  12. )
  13. // routingServer is an implementation of RoutingService.
  14. type routingServer struct {
  15. router routing.Router
  16. routingStats stats.Channel
  17. }
  18. // NewRoutingServer creates a statistics service with statistics manager.
  19. func NewRoutingServer(router routing.Router, routingStats stats.Channel) RoutingServiceServer {
  20. return &routingServer{
  21. router: router,
  22. routingStats: routingStats,
  23. }
  24. }
  25. func (s *routingServer) TestRoute(ctx context.Context, request *TestRouteRequest) (*RoutingContext, error) {
  26. if request.RoutingContext == nil {
  27. return nil, newError("Invalid routing request.")
  28. }
  29. route, err := s.router.PickRoute(AsRoutingContext(request.RoutingContext))
  30. if err != nil {
  31. return nil, err
  32. }
  33. if request.PublishResult && s.routingStats != nil {
  34. ctx, _ := context.WithTimeout(context.Background(), 4*time.Second)
  35. s.routingStats.Publish(ctx, route)
  36. }
  37. return AsProtobufMessage(request.FieldSelectors)(route), nil
  38. }
  39. func (s *routingServer) SubscribeRoutingStats(request *SubscribeRoutingStatsRequest, stream RoutingService_SubscribeRoutingStatsServer) error {
  40. if s.routingStats == nil {
  41. return newError("Routing statistics not enabled.")
  42. }
  43. genMessage := AsProtobufMessage(request.FieldSelectors)
  44. subscriber, err := stats.SubscribeRunnableChannel(s.routingStats)
  45. if err != nil {
  46. return err
  47. }
  48. defer stats.UnsubscribeClosableChannel(s.routingStats, subscriber)
  49. for {
  50. select {
  51. case value, ok := <-subscriber:
  52. if !ok {
  53. return newError("Upstream closed the subscriber channel.")
  54. }
  55. route, ok := value.(routing.Route)
  56. if !ok {
  57. return newError("Upstream sent malformed statistics.")
  58. }
  59. err := stream.Send(genMessage(route))
  60. if err != nil {
  61. return err
  62. }
  63. case <-stream.Context().Done():
  64. return stream.Context().Err()
  65. }
  66. }
  67. }
  68. func (s *routingServer) mustEmbedUnimplementedRoutingServiceServer() {}
  69. type service struct {
  70. v *core.Instance
  71. }
  72. func (s *service) Register(server *grpc.Server) {
  73. common.Must(s.v.RequireFeatures(func(router routing.Router, stats stats.Manager) {
  74. rs := NewRoutingServer(router, nil)
  75. RegisterRoutingServiceServer(server, rs)
  76. // For compatibility purposes
  77. vCoreDesc := _RoutingService_serviceDesc
  78. vCoreDesc.ServiceName = "v2ray.core.app.router.command.RoutingService"
  79. server.RegisterService(&vCoreDesc, rs)
  80. }))
  81. }
  82. func init() {
  83. common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, cfg interface{}) (interface{}, error) {
  84. s := core.MustFromContext(ctx)
  85. return &service{v: s}, nil
  86. }))
  87. }