add deleteTopic, refactoring

This commit is contained in:
Chris Lu 2020-05-09 00:31:34 -07:00
parent dfccc3c263
commit 07d7abe428
4 changed files with 83 additions and 56 deletions

View file

@ -2,7 +2,10 @@ package broker
import ( import (
"context" "context"
"fmt"
"github.com/chrislusf/seaweedfs/weed/filer2"
"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb" "github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
) )
@ -11,9 +14,24 @@ func (broker *MessageBroker) ConfigureTopic(c context.Context, request *messagin
} }
func (broker *MessageBroker) DeleteTopic(c context.Context, request *messaging_pb.DeleteTopicRequest) (*messaging_pb.DeleteTopicResponse, error) { func (broker *MessageBroker) DeleteTopic(c context.Context, request *messaging_pb.DeleteTopicRequest) (*messaging_pb.DeleteTopicResponse, error) {
panic("implement me") 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) { func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error) {
panic("implement me") 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/%s", filer2.TopicsDir, namespace), topic
}

View file

@ -119,7 +119,7 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
sizeBuf := make([]byte, 4) sizeBuf := make([]byte, 4)
startTsNs := startTime.UnixNano() startTsNs := startTime.UnixNano()
topicDir := fmt.Sprintf("/topics/%s/%s", tp.Namespace, tp.Topic) topicDir := genTopicDir(tp.Namespace, tp.Topic)
partitionSuffix := fmt.Sprintf(".part%02d", tp.Partition) partitionSuffix := fmt.Sprintf(".part%02d", tp.Partition)
return filer_pb.List(broker, topicDir, "", func(dayEntry *filer_pb.Entry, isLast bool) error { return filer_pb.List(broker, topicDir, "", func(dayEntry *filer_pb.Entry, isLast bool) error {

View file

@ -3,7 +3,6 @@ package msgclient
import ( import (
"io" "io"
"log" "log"
"time"
"google.golang.org/grpc" "google.golang.org/grpc"
@ -63,56 +62,3 @@ func (pc *PubChannel) Close() error {
} }
return nil return nil
} }
type SubChannel struct {
ch chan []byte
stream messaging_pb.SeaweedMessaging_SubscribeClient
}
func (mc *MessagingClient) NewSubChannel(chanName string) (*SubChannel, error) {
tp := broker.TopicPartition{
Namespace: "chan",
Topic: chanName,
Partition: 0,
}
grpcConnection, err := mc.findBroker(tp)
if err != nil {
return nil, err
}
sc, err := setupSubscriberClient(grpcConnection, "", "chan", chanName, 0, time.Unix(0, 0))
if err != nil {
return nil, err
}
t := &SubChannel{
ch: make(chan []byte),
stream: sc,
}
go func() {
for {
resp, subErr := t.stream.Recv()
if subErr == io.EOF {
return
}
if subErr != nil {
log.Printf("fail to receive from netchan %s: %v", chanName, subErr)
return
}
if resp.Data.IsClose {
t.stream.Send(&messaging_pb.SubscriberMessage{
IsClose: true,
})
close(t.ch)
return
}
t.ch <- resp.Data.Value
}
}()
return t, nil
}
func (sc *SubChannel) Channel() chan []byte {
return sc.ch
}

View file

@ -0,0 +1,63 @@
package msgclient
import (
"io"
"log"
"time"
"github.com/chrislusf/seaweedfs/weed/messaging/broker"
"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
)
type SubChannel struct {
ch chan []byte
stream messaging_pb.SeaweedMessaging_SubscribeClient
}
func (mc *MessagingClient) NewSubChannel(chanName string) (*SubChannel, error) {
tp := broker.TopicPartition{
Namespace: "chan",
Topic: chanName,
Partition: 0,
}
grpcConnection, err := mc.findBroker(tp)
if err != nil {
return nil, err
}
sc, err := setupSubscriberClient(grpcConnection, "", "chan", chanName, 0, time.Unix(0, 0))
if err != nil {
return nil, err
}
t := &SubChannel{
ch: make(chan []byte),
stream: sc,
}
go func() {
for {
resp, subErr := t.stream.Recv()
if subErr == io.EOF {
return
}
if subErr != nil {
log.Printf("fail to receive from netchan %s: %v", chanName, subErr)
return
}
if resp.Data.IsClose {
t.stream.Send(&messaging_pb.SubscriberMessage{
IsClose: true,
})
close(t.ch)
return
}
t.ch <- resp.Data.Value
}
}()
return t, nil
}
func (sc *SubChannel) Channel() chan []byte {
return sc.ch
}