gitlab-com-seaweedfs/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2024-01-19 02:48:24 +00:00
Code Issues Projects Releases Wiki Activity

broker: read cipher value from filer

Browse source
This commit is contained in:
Chris Lu 2020-04-17 02:29:00 -07:00
parent bda82f61bc
commit 2a45897237
2 changed files with 11 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
weed/command/msg_broker.go
View file

@ -22,8 +22,8 @@ var (
) )
type QueueOptions struct { type QueueOptions struct {
filer *string filer *string
port *int port *int
} }
func init() { func init() {
@ -59,14 +59,16 @@ func (msgBrokerOpt *QueueOptions) startQueueServer() bool {
return false return false
} }
grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client") grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.msg_broker")
cipher := false
for { for {
err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { err = pb.WithGrpcFilerClient(filerGrpcAddress, grpcDialOption, func(client filer_pb.SeaweedFilerClient) error {
_, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{}) resp, err := client.GetFilerConfiguration(context.Background(), &filer_pb.GetFilerConfigurationRequest{})
if err != nil { if err != nil {
return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err) return fmt.Errorf("get filer %s configuration: %v", filerGrpcAddress, err)
} }
cipher = resp.Cipher
return nil return nil
}) })
if err != nil { if err != nil {
@ -83,7 +85,8 @@ func (msgBrokerOpt *QueueOptions) startQueueServer() bool {
DefaultReplication: "", DefaultReplication: "",
MaxMB: 0, MaxMB: 0,
Port: *msgBrokerOpt.port, Port: *msgBrokerOpt.port,
}) Cipher: cipher,
}, grpcDialOption)
// start grpc listener // start grpc listener
grpcL, err := util.NewListener(":"+strconv.Itoa(*msgBrokerOpt.port), 0) grpcL, err := util.NewListener(":"+strconv.Itoa(*msgBrokerOpt.port), 0)

7
weed/messaging/msg_broker_server.go
View file

@ -10,8 +10,6 @@ import (
"github.com/chrislusf/seaweedfs/weed/pb" "github.com/chrislusf/seaweedfs/weed/pb"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb" "github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/master_pb" "github.com/chrislusf/seaweedfs/weed/pb/master_pb"
"github.com/chrislusf/seaweedfs/weed/security"
"github.com/chrislusf/seaweedfs/weed/util"
) )
type MessageBrokerOption struct { type MessageBrokerOption struct {
@ -19,6 +17,7 @@ type MessageBrokerOption struct {
DefaultReplication string DefaultReplication string
MaxMB int MaxMB int
Port int Port int
Cipher bool
} }
type MessageBroker struct { type MessageBroker struct {
@ -26,11 +25,11 @@ type MessageBroker struct {
grpcDialOption grpc.DialOption grpcDialOption grpc.DialOption
} }
func NewMessageBroker(option *MessageBrokerOption) (messageBroker *MessageBroker, err error) { func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOption) (messageBroker *MessageBroker, err error) {
messageBroker = &MessageBroker{ messageBroker = &MessageBroker{
option: option, option: option,
grpcDialOption: security.LoadClientTLS(util.GetViper(), "grpc.msg_broker"), grpcDialOption: grpcDialOption,
} }
go messageBroker.loopForEver() go messageBroker.loopForEver()