Merge pull request #1318 from chrislusf/msg_channel

Add messaging, add channel

1 files changed, 22 insertions, 0 deletions

diff --git a/weed/messaging/broker/broker_grpc_server.go b/weed/messaging/broker/broker_grpc_server.go
index 447620a6b..6918a28a6 100644
--- a/weed/messaging/broker/broker_grpc_server.go
+++ b/weed/messaging/broker/broker_grpc_server.go
@@ -2,7 +2,10 @@ package broker
 
 import (
 	"context"
+	"fmt"
 
+	"github.com/chrislusf/seaweedfs/weed/filer2"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
 )
 
@@ -10,6 +13,25 @@ func (broker *MessageBroker) ConfigureTopic(c context.Context, request *messagin
 	panic("implement me")
 }
 
+func (broker *MessageBroker) DeleteTopic(c context.Context, request *messaging_pb.DeleteTopicRequest) (*messaging_pb.DeleteTopicResponse, error) {
+	resp := &messaging_pb.DeleteTopicResponse{}
+	dir, entry := genTopicDirEntry(request.Namespace, request.Topic)
+	if exists, err := filer_pb.Exists(broker, dir, entry, true); err != nil {
+		return nil, err
+	} else if exists {
+		err = filer_pb.Remove(broker, dir, entry, true, true, true)
+	}
+	return resp, nil
+}
+
 func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error) {
 	panic("implement me")
 }
+
+func genTopicDir(namespace, topic string) string {
+	return fmt.Sprintf("%s/%s/%s", filer2.TopicsDir, namespace, topic)
+}
+
+func genTopicDirEntry(namespace, topic string) (dir, entry string) {
+	return fmt.Sprintf("%s/%s", filer2.TopicsDir, namespace), topic
+}