1 files changed, 39 insertions, 5 deletions
diff --git a/weed/mq/topic/topic.go b/weed/mq/topic/topic.go
index 6932fcb56..5e9012e70 100644
--- a/weed/mq/topic/topic.go
+++ b/weed/mq/topic/topic.go
@@ -1,8 +1,13 @@
 package topic
 
 import (
+	"bytes"
+	"errors"
 	"fmt"
+	"github.com/seaweedfs/seaweedfs/weed/filer"
+	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
 	"github.com/seaweedfs/seaweedfs/weed/pb/mq_pb"
+	jsonpb "google.golang.org/protobuf/encoding/protojson"
 )
 
 type Topic struct {
@@ -23,13 +28,42 @@ func FromPbTopic(topic *mq_pb.Topic) Topic {
 	}
 }
 
-func (tp Topic) ToPbTopic() *mq_pb.Topic {
+func (t Topic) ToPbTopic() *mq_pb.Topic {
 	return &mq_pb.Topic{
-		Namespace: tp.Namespace,
-		Name:      tp.Name,
+		Namespace: t.Namespace,
+		Name:      t.Name,
 	}
 }
 
-func (tp Topic) String() string {
-	return fmt.Sprintf("%s.%s", tp.Namespace, tp.Name)
+func (t Topic) String() string {
+	return fmt.Sprintf("%s.%s", t.Namespace, t.Name)
+}
+
+func (t Topic) Dir() string {
+	return fmt.Sprintf("%s/%s/%s", filer.TopicsDir, t.Namespace, t.Name)
+}
+
+func (t Topic) ReadConfFile(client filer_pb.SeaweedFilerClient) (*mq_pb.ConfigureTopicResponse, error) {
+	data, err := filer.ReadInsideFiler(client, t.Dir(), filer.TopicConfFile)
+	if errors.Is(err, filer_pb.ErrNotFound) {
+		return nil, err
+	}
+	if err != nil {
+		return nil, fmt.Errorf("read topic.conf of %v: %v", t, err)
+	}
+	// parse into filer conf object
+	conf := &mq_pb.ConfigureTopicResponse{}
+	if err = jsonpb.Unmarshal(data, conf); err != nil {
+		return nil, fmt.Errorf("unmarshal topic %v conf: %v", t, err)
+	}
+	return conf, nil
+}
+
+func (t Topic) WriteConfFile(client filer_pb.SeaweedFilerClient, conf *mq_pb.ConfigureTopicResponse) error {
+	var buf bytes.Buffer
+	filer.ProtoToText(&buf, conf)
+	if err := filer.SaveInsideFiler(client, t.Dir(), filer.TopicConfFile, buf.Bytes()); err != nil {
+		return fmt.Errorf("save topic %v conf: %v", t, err)
+	}
+	return nil
 }