commander.go 2.2 KB

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