You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

110 lines
3.3 KiB

5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
5 years ago
  1. package command
  2. import (
  3. "context"
  4. "fmt"
  5. "strconv"
  6. "time"
  7. "github.com/chrislusf/seaweedfs/weed/util/grace"
  8. "google.golang.org/grpc/reflection"
  9. "github.com/chrislusf/seaweedfs/weed/glog"
  10. "github.com/chrislusf/seaweedfs/weed/messaging/broker"
  11. "github.com/chrislusf/seaweedfs/weed/pb"
  12. "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
  13. "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
  14. "github.com/chrislusf/seaweedfs/weed/security"
  15. "github.com/chrislusf/seaweedfs/weed/util"
  16. )
  17. var (
  18. messageBrokerStandaloneOptions QueueOptions
  19. )
  20. type QueueOptions struct {
  21. filer *string
  22. port *int
  23. cpuprofile *string
  24. memprofile *string
  25. }
  26. func init() {
  27. cmdMsgBroker.Run = runMsgBroker // break init cycle
  28. messageBrokerStandaloneOptions.filer = cmdMsgBroker.Flag.String("filer", "localhost:8888", "filer server address")
  29. messageBrokerStandaloneOptions.port = cmdMsgBroker.Flag.Int("port", 17777, "queue server gRPC listen port")
  30. messageBrokerStandaloneOptions.cpuprofile = cmdMsgBroker.Flag.String("cpuprofile", "", "cpu profile output file")
  31. messageBrokerStandaloneOptions.memprofile = cmdMsgBroker.Flag.String("memprofile", "", "memory profile output file")
  32. }
  33. var cmdMsgBroker = &Command{
  34. UsageLine: "msg.broker [-port=17777] [-filer=<ip:port>]",
  35. Short: "<WIP> start a message queue broker",
  36. Long: `start a message queue broker
  37. The broker can accept gRPC calls to write or read messages. The messages are stored via filer.
  38. The brokers are stateless. To scale up, just add more brokers.
  39. `,
  40. }
  41. func runMsgBroker(cmd *Command, args []string) bool {
  42. util.LoadConfiguration("security", false)
  43. return messageBrokerStandaloneOptions.startQueueServer()
  44. }
  45. func (msgBrokerOpt *QueueOptions) startQueueServer() bool {
  46. grace.SetupProfiling(*messageBrokerStandaloneOptions.cpuprofile, *messageBrokerStandaloneOptions.memprofile)
  47. filerGrpcAddress, err := pb.ParseFilerGrpcAddress(*msgBrokerOpt.filer)
  48. if err != nil {
  49. glog.Fatal(err)
  50. return false
  51. }
  52. grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.msg_broker")
  53. cipher := false
  54. for {
  55. err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
  56. resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
  57. if err != nil {
  58. return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
  59. }
  60. cipher = resp.Cipher
  61. return nil
  62. })
  63. if err != nil {
  64. glog.V(0).Infof("wait to connect to filer %s grpc address %s", *msgBrokerOpt.filer, filerGrpcAddress)
  65. time.Sleep(time.Second)
  66. } else {
  67. glog.V(0).Infof("connected to filer %s grpc address %s", *msgBrokerOpt.filer, filerGrpcAddress)
  68. break
  69. }
  70. }
  71. qs, err := broker.NewMessageBroker(&broker.MessageBrokerOption{
  72. Filers: []string{*msgBrokerOpt.filer},
  73. DefaultReplication: "",
  74. MaxMB: 0,
  75. Port: *msgBrokerOpt.port,
  76. Cipher: cipher,
  77. }, grpcDialOption)
  78. // start grpc listener
  79. grpcL, err := util.NewListener(":"+strconv.Itoa(*msgBrokerOpt.port), 0)
  80. if err != nil {
  81. glog.Fatalf("failed to listen on grpc port %d: %v", *msgBrokerOpt.port, err)
  82. }
  83. grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.msg_broker"))
  84. messaging_pb.RegisterSeaweedMessagingServer(grpcS, qs)
  85. reflection.Register(grpcS)
  86. grpcS.Serve(grpcL)
  87. return true
  88. }