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.FromContext(ctx)
  21. if v == nil {
  22. return nil, newError("V is not in context.")
  23. }
  24. c := &Commander{
  25. config: *config,
  26. ohm: v.OutboundHandlerManager(),
  27. v: v,
  28. }
  29. if err := v.RegisterFeature((*Commander)(nil), c); err != nil {
  30. return nil, err
  31. }
  32. return c, nil
  33. }
  34. func (c *Commander) Type() interface{} {
  35. return (*Commander)(nil)
  36. }
  37. func (c *Commander) Start() error {
  38. c.Lock()
  39. c.server = grpc.NewServer()
  40. for _, rawConfig := range c.config.Service {
  41. config, err := rawConfig.GetInstance()
  42. if err != nil {
  43. return err
  44. }
  45. rawService, err := c.v.CreateObject(config)
  46. if err != nil {
  47. return err
  48. }
  49. service, ok := rawService.(Service)
  50. if !ok {
  51. return newError("not a Service.")
  52. }
  53. service.Register(c.server)
  54. }
  55. c.Unlock()
  56. listener := &OutboundListener{
  57. buffer: make(chan net.Conn, 4),
  58. done: signal.NewDone(),
  59. }
  60. go func() {
  61. if err := c.server.Serve(listener); err != nil {
  62. newError("failed to start grpc server").Base(err).AtError().WriteToLog()
  63. }
  64. }()
  65. c.ohm.RemoveHandler(context.Background(), c.config.Tag)
  66. c.ohm.AddHandler(context.Background(), &CommanderOutbound{
  67. tag: c.config.Tag,
  68. listener: listener,
  69. })
  70. return nil
  71. }
  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. }