commander.go 2.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110
  1. // +build !confonly
  2. package commander
  3. //go:generate errorgen
  4. import (
  5. "context"
  6. "net"
  7. "sync"
  8. "google.golang.org/grpc"
  9. "v2ray.com/core"
  10. "v2ray.com/core/common"
  11. "v2ray.com/core/common/signal/done"
  12. "v2ray.com/core/features/outbound"
  13. )
  14. // Commander is a V2Ray feature that provides gRPC methods to external clients.
  15. type Commander struct {
  16. sync.Mutex
  17. server *grpc.Server
  18. services []Service
  19. ohm outbound.Manager
  20. tag string
  21. }
  22. // NewCommander creates a new Commander based on the given config.
  23. func NewCommander(ctx context.Context, config *Config) (*Commander, error) {
  24. c := &Commander{
  25. tag: config.Tag,
  26. }
  27. common.Must(core.RequireFeatures(ctx, func(om outbound.Manager) {
  28. c.ohm = om
  29. }))
  30. for _, rawConfig := range config.Service {
  31. config, err := rawConfig.GetInstance()
  32. if err != nil {
  33. return nil, err
  34. }
  35. rawService, err := common.CreateObject(ctx, config)
  36. if err != nil {
  37. return nil, err
  38. }
  39. service, ok := rawService.(Service)
  40. if !ok {
  41. return nil, newError("not a Service.")
  42. }
  43. c.services = append(c.services, service)
  44. }
  45. return c, nil
  46. }
  47. // Type implements common.HasType.
  48. func (c *Commander) Type() interface{} {
  49. return (*Commander)(nil)
  50. }
  51. // Start implements common.Runnable.
  52. func (c *Commander) Start() error {
  53. c.Lock()
  54. c.server = grpc.NewServer()
  55. for _, service := range c.services {
  56. service.Register(c.server)
  57. }
  58. c.Unlock()
  59. listener := &OutboundListener{
  60. buffer: make(chan net.Conn, 4),
  61. done: done.New(),
  62. }
  63. go func() {
  64. if err := c.server.Serve(listener); err != nil {
  65. newError("failed to start grpc server").Base(err).AtError().WriteToLog()
  66. }
  67. }()
  68. if err := c.ohm.RemoveHandler(context.Background(), c.tag); err != nil {
  69. newError("failed to remove existing handler").WriteToLog()
  70. }
  71. return c.ohm.AddHandler(context.Background(), &Outbound{
  72. tag: c.tag,
  73. listener: listener,
  74. })
  75. }
  76. // Close implements common.Closable.
  77. func (c *Commander) Close() error {
  78. c.Lock()
  79. defer c.Unlock()
  80. if c.server != nil {
  81. c.server.Stop()
  82. c.server = nil
  83. }
  84. return nil
  85. }
  86. func init() {
  87. common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, cfg interface{}) (interface{}, error) {
  88. return NewCommander(ctx, cfg.(*Config))
  89. }))
  90. }