seaweedfs/weed/command/mq_broker.go

110 lines
3.4 KiB
Go
Raw Normal View History

2020-03-04 08:39:47 +00:00
package command
import (
"context"
"fmt"
"time"
"google.golang.org/grpc/reflection"
2020-05-05 09:05:28 +00:00
"github.com/chrislusf/seaweedfs/weed/util/grace"
2020-04-16 09:55:09 +00:00
"github.com/chrislusf/seaweedfs/weed/glog"
2022-07-02 05:43:25 +00:00
"github.com/chrislusf/seaweedfs/weed/mq/broker"
2020-03-04 08:39:47 +00:00
"github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
2022-07-02 05:43:25 +00:00
"github.com/chrislusf/seaweedfs/weed/pb/mq_pb"
2020-03-04 08:39:47 +00:00
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
)
var (
2022-07-02 04:59:50 +00:00
mqBrokerStandaloneOptions MessageQueueBrokerOptions
2020-03-04 08:39:47 +00:00
)
2022-07-02 04:59:50 +00:00
type MessageQueueBrokerOptions struct {
2020-04-19 10:03:40 +00:00
filer *string
2020-05-05 09:05:28 +00:00
ip *string
2020-04-19 10:03:40 +00:00
port *int
cpuprofile *string
memprofile *string
2020-03-04 08:39:47 +00:00
}
func init() {
2022-07-02 04:59:50 +00:00
cmdMqBroker.Run = runMqBroker // break init cycle
mqBrokerStandaloneOptions.filer = cmdMqBroker.Flag.String("filer", "localhost:8888", "filer server address")
mqBrokerStandaloneOptions.ip = cmdMqBroker.Flag.String("ip", util.DetectedHostAddress(), "broker host address")
mqBrokerStandaloneOptions.port = cmdMqBroker.Flag.Int("port", 17777, "broker gRPC listen port")
mqBrokerStandaloneOptions.cpuprofile = cmdMqBroker.Flag.String("cpuprofile", "", "cpu profile output file")
mqBrokerStandaloneOptions.memprofile = cmdMqBroker.Flag.String("memprofile", "", "memory profile output file")
2020-03-04 08:39:47 +00:00
}
2022-07-02 04:59:50 +00:00
var cmdMqBroker = &Command{
UsageLine: "mq.broker [-port=17777] [-filer=<ip:port>]",
Short: "start a message queue broker",
2020-03-04 08:39:47 +00:00
Long: `start a message queue broker
The broker can accept gRPC calls to write or read messages. The messages are stored via filer.
The brokers are stateless. To scale up, just add more brokers.
`,
}
2022-07-02 04:59:50 +00:00
func runMqBroker(cmd *Command, args []string) bool {
2020-03-04 08:39:47 +00:00
util.LoadConfiguration("security", false)
2022-07-02 04:59:50 +00:00
return mqBrokerStandaloneOptions.startQueueServer()
2020-03-04 08:39:47 +00:00
}
2022-07-02 04:59:50 +00:00
func (mqBrokerOpt *MessageQueueBrokerOptions) startQueueServer() bool {
2020-03-04 08:39:47 +00:00
2022-07-02 04:59:50 +00:00
grace.SetupProfiling(*mqBrokerStandaloneOptions.cpuprofile, *mqBrokerStandaloneOptions.memprofile)
2020-04-19 10:03:40 +00:00
2022-07-02 04:59:50 +00:00
filerAddress := pb.ServerAddress(*mqBrokerOpt.filer)
2020-03-04 08:39:47 +00:00
2020-04-17 09:29:00 +00:00
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.msg_broker")
cipher := false
2020-03-04 08:39:47 +00:00
for {
err := pb.WithGrpcFilerClient(false, filerAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
2020-04-17 09:29:00 +00:00
resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
2020-03-04 08:39:47 +00:00
if err != nil {
return fmt.Errorf("get filer %s configuration: %v", filerAddress, err)
2020-03-04 08:39:47 +00:00
}
2020-04-17 09:29:00 +00:00
cipher = resp.Cipher
2020-03-04 08:39:47 +00:00
return nil
})
if err != nil {
2022-07-02 04:59:50 +00:00
glog.V(0).Infof("wait to connect to filer %s grpc address %s", *mqBrokerOpt.filer, filerAddress.ToGrpcAddress())
2020-03-04 08:39:47 +00:00
time.Sleep(time.Second)
} else {
2022-07-02 04:59:50 +00:00
glog.V(0).Infof("connected to filer %s grpc address %s", *mqBrokerOpt.filer, filerAddress.ToGrpcAddress())
2020-03-04 08:39:47 +00:00
break
}
}
2020-04-18 08:12:01 +00:00
qs, err := broker.NewMessageBroker(&broker.MessageBrokerOption{
Filers: []pb.ServerAddress{filerAddress},
2020-03-04 08:39:47 +00:00
DefaultReplication: "",
MaxMB: 0,
2022-07-02 04:59:50 +00:00
Ip: *mqBrokerOpt.ip,
Port: *mqBrokerOpt.port,
2020-04-17 09:29:00 +00:00
Cipher: cipher,
}, grpcDialOption)
2020-03-04 08:39:47 +00:00
// start grpc listener
2022-07-02 04:59:50 +00:00
grpcL, _, err := util.NewIpAndLocalListeners("", *mqBrokerOpt.port, 0)
2020-03-04 08:39:47 +00:00
if err != nil {
2022-07-02 04:59:50 +00:00
glog.Fatalf("failed to listen on grpc port %d: %v", *mqBrokerOpt.port, err)
2020-03-04 08:39:47 +00:00
}
grpcS := pb.NewGrpcServer(security.LoadServerTLS(util.GetViper(), "grpc.msg_broker"))
2022-07-02 05:43:25 +00:00
mq_pb.RegisterSeaweedMessagingServer(grpcS, qs)
2020-03-04 08:39:47 +00:00
reflection.Register(grpcS)
grpcS.Serve(grpcL)
return true
}