commander.go 2.1 KB

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