mirror/seaweedfs
1
0
Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2025-01-19 06:53:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor

Browse Source
This commit is contained in:
chrislu 2024-01-21 01:23:43 -08:00
parent 34839237ab
commit a5b906c0dd
3 changed files with 52 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
weed/mq/broker/broker_grpc_configure.go
View File

@ -1,7 +1,6 @@
package broker
import (
"bytes"
"context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
@ -50,13 +49,8 @@ func (b *MessageQueueBroker) ConfigureTopic(ctx context.Context, request *mq_pb.
ret.BrokerPartitionAssignments = pub_balancer.AllocateTopicPartitions(b.Balancer.Brokers, request.PartitionCount)
// save the topic configuration on filer
topicDir := fmt.Sprintf("%s/%s/%s", filer.TopicsDir, request.Topic.Namespace, request.Topic.Name)
if err = b.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
var buf bytes.Buffer
filer.ProtoToText(&buf, ret)
return filer.SaveInsideFiler(client, topicDir, "topic.conf", buf.Bytes())
}); err != nil {
return nil, fmt.Errorf("create topic %s: %v", topicDir, err)
if err := b.saveTopicConfToFiler(request.Topic, ret); err != nil {
return nil, fmt.Errorf("configure topic: %v", err)
}
b.Balancer.OnPartitionChange(request.Topic, ret.BrokerPartitionAssignments)

25
weed/mq/broker/broker_grpc_pub.go
View File

@ -3,13 +3,10 @@ package broker
import (
"context"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
"google.golang.org/grpc/peer"
jsonpb "google.golang.org/protobuf/encoding/protojson"
"math/rand"
"net"
"sync/atomic"
@ -180,28 +177,6 @@ func (b *MessageQueueBroker) loadLocalTopicPartitionFromFiler(t topic.Topic, p t
return localTopicPartition, nil
}
func (b *MessageQueueBroker) readTopicConfFromFiler(t topic.Topic) (conf *mq_pb.ConfigureTopicResponse, err error) {
glog.V(0).Infof("load conf for topic %v from filer", t)
topicDir := fmt.Sprintf("%s/%s/%s", filer.TopicsDir, t.Namespace, t.Name)
if err = b.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := filer.ReadInsideFiler(client, topicDir, "topic.conf")
if err != nil {
return fmt.Errorf("read topic %v partition %v conf: %v", t, err)
}
// parse into filer conf object
conf = &mq_pb.ConfigureTopicResponse{}
if err = jsonpb.Unmarshal(data, conf); err != nil {
return fmt.Errorf("unmarshal topic %v conf: %v", t, err)
}
return nil
}); err != nil {
return nil, err
}
return conf, err
}
// duplicated from master_grpc_server.go
func findClientAddress(ctx context.Context) string {
// fmt.Printf("FromContext %+v\n", ctx)

50
weed/mq/broker/broker_topic_conf_read_write.go Normal file
View File

@ -0,0 +1,50 @@
package broker
import (
"bytes"
"fmt"
"github.com/seaweedfs/seaweedfs/weed/filer"
"github.com/seaweedfs/seaweedfs/weed/glog"
"github.com/seaweedfs/seaweedfs/weed/mq/topic"
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
jsonpb "google.golang.org/protobuf/encoding/protojson"
)
func (b *MessageQueueBroker) saveTopicConfToFiler(t *mq_pb.Topic, conf *mq_pb.ConfigureTopicResponse) error {
glog.V(0).Infof("save conf for topic %v to filer", t)
// save the topic configuration on filer
topicDir := fmt.Sprintf("%s/%s/%s", filer.TopicsDir, t.Namespace, t.Name)
if err := b.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
var buf bytes.Buffer
filer.ProtoToText(&buf, conf)
return filer.SaveInsideFiler(client, topicDir, "topic.conf", buf.Bytes())
}); err != nil {
return fmt.Errorf("save topic to %s: %v", topicDir, err)
}
return nil
}
func (b *MessageQueueBroker) readTopicConfFromFiler(t topic.Topic) (conf *mq_pb.ConfigureTopicResponse, err error) {
glog.V(0).Infof("load conf for topic %v from filer", t)
topicDir := fmt.Sprintf("%s/%s/%s", filer.TopicsDir, t.Namespace, t.Name)
if err = b.WithFilerClient(false, func(client filer_pb.SeaweedFilerClient) error {
data, err := filer.ReadInsideFiler(client, topicDir, "topic.conf")
if err != nil {
return fmt.Errorf("read topic %v partition %v conf: %v", t, err)
}
// parse into filer conf object
conf = &mq_pb.ConfigureTopicResponse{}
if err = jsonpb.Unmarshal(data, conf); err != nil {
return fmt.Errorf("unmarshal topic %v conf: %v", t, err)
}
return nil
}); err != nil {
return nil, err
}
return conf, err
}