commander.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899
  1. package commander
  2. //go:generate go run $GOPATH/src/v2ray.com/core/common/errors/errorgen/main.go -pkg commander -path App,Commander
  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"
  11. )
  12. type Commander struct {
  13. sync.Mutex
  14. server *grpc.Server
  15. config Config
  16. v *core.Instance
  17. ohm core.OutboundHandlerManager
  18. }
  19. func NewCommander(ctx context.Context, config *Config) (*Commander, error) {
  20. v := core.MustFromContext(ctx)
  21. c := &Commander{
  22. config: *config,
  23. ohm: v.OutboundHandlerManager(),
  24. v: v,
  25. }
  26. if err := v.RegisterFeature((*Commander)(nil), c); err != nil {
  27. return nil, err
  28. }
  29. return c, nil
  30. }
  31. // Type implements common.HasType.
  32. func (c *Commander) Type() interface{} {
  33. return (*Commander)(nil)
  34. }
  35. // Start implements common.Runnable.
  36. func (c *Commander) Start() error {
  37. c.Lock()
  38. c.server = grpc.NewServer()
  39. for _, rawConfig := range c.config.Service {
  40. config, err := rawConfig.GetInstance()
  41. if err != nil {
  42. return err
  43. }
  44. rawService, err := c.v.CreateObject(config)
  45. if err != nil {
  46. return err
  47. }
  48. service, ok := rawService.(Service)
  49. if !ok {
  50. return newError("not a Service.")
  51. }
  52. service.Register(c.server)
  53. }
  54. c.Unlock()
  55. listener := &OutboundListener{
  56. buffer: make(chan net.Conn, 4),
  57. done: signal.NewDone(),
  58. }
  59. go func() {
  60. if err := c.server.Serve(listener); err != nil {
  61. newError("failed to start grpc server").Base(err).AtError().WriteToLog()
  62. }
  63. }()
  64. c.ohm.RemoveHandler(context.Background(), c.config.Tag)
  65. c.ohm.AddHandler(context.Background(), &CommanderOutbound{
  66. tag: c.config.Tag,
  67. listener: listener,
  68. })
  69. return nil
  70. }
  71. // Close implements common.Closable.
  72. func (c *Commander) Close() error {
  73. c.Lock()
  74. defer c.Unlock()
  75. if c.server != nil {
  76. c.server.Stop()
  77. c.server = nil
  78. }
  79. return nil
  80. }
  81. func init() {
  82. common.Must(common.RegisterConfig((*Config)(nil), func(ctx context.Context, cfg interface{}) (interface{}, error) {
  83. return NewCommander(ctx, cfg.(*Config))
  84. }))
  85. }