able to read chan and write chan

weed/messaging/broker/broker_append.go

@@ -3,6 +3,7 @@ package broker
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
 	"github.com/chrislusf/seaweedfs/weed/operation"
@@ -94,6 +95,9 @@ func (broker *MessageBroker) WithFilerClient(fn func(filer_pb.SeaweedFilerClient
 
 	for _, filer := range broker.option.Filers {
 		if err = pb.WithFilerClient(filer, broker.grpcDialOption, fn); err != nil {
+			if err == io.EOF {
+				return
+			}
 			glog.V(0).Infof("fail to connect to %s: %v", filer, err)
 		} else {
 			break

weed/messaging/broker/broker_grpc_server.go

@@ -10,6 +10,10 @@ 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) {
+	panic("implement me")
+}
+
 func (broker *MessageBroker) GetTopicConfiguration(c context.Context, request *messaging_pb.GetTopicConfigurationRequest) (*messaging_pb.GetTopicConfigurationResponse, error) {
 	panic("implement me")
 }

weed/messaging/broker/broker_grpc_server_discovery.go

@@ -2,6 +2,7 @@ package broker
 
 import (
 	"context"
+	"fmt"
 	"time"
 
 	"github.com/chrislusf/seaweedfs/weed/glog"
@@ -25,12 +26,40 @@ If one of the pub or sub connects very late, and the system topo changed quite a
 
 func (broker *MessageBroker) FindBroker(c context.Context, request *messaging_pb.FindBrokerRequest) (*messaging_pb.FindBrokerResponse, error) {
 
-	panic("implement me")
+	t := &messaging_pb.FindBrokerResponse{}
+	var peers []string
+
+	targetTopicPartition := fmt.Sprintf(TopicPartitionFmt, request.Namespace, request.Topic, request.Parition)
+
+	for _, filer := range broker.option.Filers {
+		err := broker.withFilerClient(filer, func(client filer_pb.SeaweedFilerClient) error {
+			resp, err := client.LocateBroker(context.Background(), &filer_pb.LocateBrokerRequest{
+				Resource: targetTopicPartition,
+			})
+			if err != nil {
+				return err
+			}
+			if resp.Found && len(resp.Resources) > 0 {
+				t.Broker = resp.Resources[0].GrpcAddresses
+				return nil
+			}
+			for _, b := range resp.Resources {
+				peers = append(peers, b.GrpcAddresses)
+			}
+			return nil
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	t.Broker = PickMember(peers, []byte(targetTopicPartition))
+
+	return t, nil
+
 }
 
-
-
-func (broker *MessageBroker) checkPeers() {
+func (broker *MessageBroker) checkFilers() {
 
 	// contact a filer about masters
 	var masters []string

weed/messaging/broker/broker_grpc_server_publish.go

@@ -47,24 +47,11 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
 	tl := broker.topicLocks.RequestLock(tp, topicConfig, true)
 	defer broker.topicLocks.ReleaseLock(tp, true)
 
-	updatesChan := make(chan int32)
-
-	go func() {
-		for update := range updatesChan {
-			if err := stream.Send(&messaging_pb.PublishResponse{
-				Config: &messaging_pb.PublishResponse_ConfigMessage{
-					PartitionCount: update,
-				},
-			}); err != nil {
-				glog.V(0).Infof("err sending publish response: %v", err)
-				return
-			}
-		}
-	}()
-
 	// process each message
 	for {
+		// println("recv")
 		in, err := stream.Recv()
+		// glog.V(0).Infof("recieved %v err: %v", in, err)
 		if err == io.EOF {
 			return nil
 		}
@@ -86,5 +73,18 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
 
 		tl.logBuffer.AddToBuffer(in.Data.Key, data)
 
+		if in.Data.IsClose {
+			// println("server received closing")
+			break
+		}
+
 	}
+
+	// send the close ack
+	// println("server send ack closing")
+	if err := stream.Send(&messaging_pb.PublishResponse{IsClosed: true}); err != nil {
+		glog.V(0).Infof("err sending close response: %v", err)
+	}
+	return nil
+
 }

weed/messaging/broker/broker_grpc_server_subscribe.go

@@ -83,11 +83,17 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 			glog.Errorf("sending %d bytes to %s: %s", len(m.Value), subscriberId, err)
 			return err
 		}
+		if m.IsClose {
+			// println("processed EOF")
+			return io.EOF
+		}
 		processedTsNs = logEntry.TsNs
+		messageCount++
 		return nil
 	}
 
 	if err := broker.readPersistedLogBuffer(&tp, lastReadTime, eachLogEntryFn); err != nil {
+		// println("stopping from persisted logs")
 		return err
 	}
 
@@ -95,7 +101,7 @@ func (broker *MessageBroker) Subscribe(stream messaging_pb.SeaweedMessaging_Subs
 		lastReadTime = time.Unix(0, processedTsNs)
 	}
 
-	messageCount, err = lock.logBuffer.LoopProcessLogData(lastReadTime, func() bool {
+	err = lock.logBuffer.LoopProcessLogData(lastReadTime, func() bool {
 		lock.Mutex.Lock()
 		lock.cond.Wait()
 		lock.Mutex.Unlock()
@@ -124,7 +130,7 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
 					return nil
 				}
 			}
-			if !strings.HasSuffix(hourMinuteEntry.Name, partitionSuffix){
+			if !strings.HasSuffix(hourMinuteEntry.Name, partitionSuffix) {
 				return nil
 			}
 			// println("partition", tp.Partition, "processing", dayDir, "/", hourMinuteEntry.Name)
@@ -133,7 +139,7 @@ func (broker *MessageBroker) readPersistedLogBuffer(tp *TopicPartition, startTim
 			if err := filer2.ReadEachLogEntry(chunkedFileReader, sizeBuf, startTsNs, eachLogEntryFn); err != nil {
 				chunkedFileReader.Close()
 				if err == io.EOF {
-					return nil
+					return err
 				}
 				return fmt.Errorf("reading %s/%s: %v", dayDir, hourMinuteEntry.Name, err)
 			}

weed/messaging/broker/broker_server.go

@@ -36,7 +36,7 @@ func NewMessageBroker(option *MessageBrokerOption, grpcDialOption grpc.DialOptio
 
 	messageBroker.topicLocks = NewTopicLocks(messageBroker)
 
-	messageBroker.checkPeers()
+	messageBroker.checkFilers()
 
 	go messageBroker.keepConnectedToOneFiler()
 
@@ -53,6 +53,24 @@ func (broker *MessageBroker) keepConnectedToOneFiler() {
 					glog.V(0).Infof("%s:%d failed to keep connected to %s: %v", broker.option.Ip, broker.option.Port, filer, err)
 					return err
 				}
+
+				initRequest := &filer_pb.KeepConnectedRequest{
+					Name:     broker.option.Ip,
+					GrpcPort: uint32(broker.option.Port),
+				}
+				for _, tp := range broker.topicLocks.ListTopicPartitions() {
+					initRequest.Resources = append(initRequest.Resources, tp.String())
+				}
+				if err := stream.Send(&filer_pb.KeepConnectedRequest{
+					Name:     broker.option.Ip,
+					GrpcPort: uint32(broker.option.Port),
+				}); err != nil {
+					glog.V(0).Infof("broker %s:%d failed to init at %s: %v", broker.option.Ip, broker.option.Port, filer, err)
+					return err
+				}
+
+				// TODO send events of adding/removing topics
+
 				glog.V(0).Infof("conntected with filer: %v", filer)
 				for {
 					if err := stream.Send(&filer_pb.KeepConnectedRequest{
@@ -68,12 +86,12 @@ func (broker *MessageBroker) keepConnectedToOneFiler() {
 						return err
 					}
 					// println("received reply")
-					time.Sleep(11*time.Second)
+					time.Sleep(11 * time.Second)
 					// println("woke up")
 				}
 				return nil
 			})
-			time.Sleep(3*time.Second)
+			time.Sleep(3 * time.Second)
 		}
 	}
 

weed/messaging/broker/consistent_distribution.go

@@ -0,0 +1,38 @@
+package broker
+
+import (
+	"github.com/cespare/xxhash"
+	"github.com/buraksezer/consistent"
+)
+
+type Member string
+
+func (m Member) String() string {
+	return string(m)
+}
+
+type hasher struct{}
+
+func (h hasher) Sum64(data []byte) uint64 {
+	return xxhash.Sum64(data)
+}
+
+func PickMember(members []string, key []byte) string {
+	cfg := consistent.Config{
+		PartitionCount:    9791,
+		ReplicationFactor: 2,
+		Load:              1.25,
+		Hasher:            hasher{},
+	}
+
+	cmembers := []consistent.Member{}
+	for _, m := range members {
+		cmembers = append(cmembers, Member(m))
+	}
+
+	c := consistent.New(cmembers, cfg)
+
+	m := c.LocateKey(key)
+
+	return m.String()
+}

weed/messaging/broker/consistent_distribution_test.go

@@ -0,0 +1,32 @@
+package broker
+
+import (
+	"fmt"
+	"testing"
+)
+
+func TestPickMember(t *testing.T) {
+
+	servers := []string{
+		"s1:port",
+		"s2:port",
+		"s3:port",
+		"s5:port",
+		"s4:port",
+	}
+
+	total := 1000
+
+	distribution := make(map[string]int)
+	for i:=0;i<total;i++{
+		tp := fmt.Sprintf("tp:%2d", i)
+		m := PickMember(servers, []byte(tp))
+		// println(tp, "=>", m)
+		distribution[m]++
+	}
+
+	for member, count := range distribution {
+		fmt.Printf("member: %s, key count: %d load=%.2f\n", member, count, float64(count*100)/float64(total/len(servers)))
+	}
+
+}

weed/messaging/broker/topic_lock.go

@@ -16,6 +16,13 @@ type TopicPartition struct {
 	Topic     string
 	Partition int32
 }
+const (
+	TopicPartitionFmt = "%s/%s_%2d"
+)
+func (tp *TopicPartition) String() string {
+	return fmt.Sprintf(TopicPartitionFmt, tp.Namespace, tp.Topic, tp.Partition)
+}
+
 type TopicLock struct {
 	sync.Mutex
 	cond            *sync.Cond
@@ -101,3 +108,13 @@ func (tl *TopicLocks) ReleaseLock(partition TopicPartition, isPublisher bool) {
 		delete(tl.locks, partition)
 	}
 }
+
+func (tl *TopicLocks) ListTopicPartitions() (tps []TopicPartition) {
+	tl.Lock()
+	defer tl.Unlock()
+
+	for k := range tl.locks {
+		tps = append(tps, k)
+	}
+	return
+}

weed/messaging/msgclient/pub_sub_chan.go

@@ -5,12 +5,15 @@ import (
 	"log"
 	"time"
 
+	"google.golang.org/grpc"
+
 	"github.com/chrislusf/seaweedfs/weed/messaging/broker"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
 )
 
 type PubChannel struct {
-	client messaging_pb.SeaweedMessaging_PublishClient
+	client         messaging_pb.SeaweedMessaging_PublishClient
+	grpcConnection *grpc.ClientConn
 }
 
 func (mc *MessagingClient) NewPubChannel(chanName string) (*PubChannel, error) {
@@ -28,7 +31,8 @@ func (mc *MessagingClient) NewPubChannel(chanName string) (*PubChannel, error) {
 		return nil, err
 	}
 	return &PubChannel{
-		client: pc,
+		client:         pc,
+		grpcConnection: grpcConnection,
 	}, nil
 }
 
@@ -40,7 +44,24 @@ func (pc *PubChannel) Publish(m []byte) error {
 	})
 }
 func (pc *PubChannel) Close() error {
-	return pc.client.CloseSend()
+
+	// println("send closing")
+	if err := pc.client.Send(&messaging_pb.PublishRequest{
+		Data: &messaging_pb.Message{
+			IsClose: true,
+		},
+	}); err != nil {
+		log.Printf("err send close: %v", err)
+	}
+	// println("receive closing")
+	if _, err := pc.client.Recv(); err != nil && err != io.EOF {
+		log.Printf("err receive close: %v", err)
+	}
+	// println("close connection")
+	if err := pc.grpcConnection.Close(); err != nil {
+		log.Printf("err connection close: %v", err)
+	}
+	return nil
 }
 
 type SubChannel struct {
@@ -58,7 +79,7 @@ func (mc *MessagingClient) NewSubChannel(chanName string) (*SubChannel, error) {
 	if err != nil {
 		return nil, err
 	}
-	sc, err := setupSubscriberClient(grpcConnection, "", "chan", chanName, 0, time.Unix(0,0))
+	sc, err := setupSubscriberClient(grpcConnection, "", "chan", chanName, 0, time.Unix(0, 0))
 	if err != nil {
 		return nil, err
 	}
@@ -78,13 +99,14 @@ func (mc *MessagingClient) NewSubChannel(chanName string) (*SubChannel, error) {
 				log.Printf("fail to receive from netchan %s: %v", chanName, subErr)
 				return
 			}
-			if resp.IsClose {
+			if resp.Data.IsClose {
+				t.stream.Send(&messaging_pb.SubscriberMessage{
+					IsClose: true,
+				})
 				close(t.ch)
 				return
 			}
-			if resp.Data != nil {
-				t.ch <- resp.Data.Value
-			}
+			t.ch <- resp.Data.Value
 		}
 	}()
 

weed/messaging/msgclient/publisher.go

@@ -4,9 +4,9 @@ import (
 	"context"
 
 	"github.com/OneOfOne/xxhash"
+	"google.golang.org/grpc"
 
 	"github.com/chrislusf/seaweedfs/weed/messaging/broker"
-	"github.com/chrislusf/seaweedfs/weed/pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
 )
 
@@ -16,7 +16,7 @@ type Publisher struct {
 	messageCount       uint64
 	publisherId        string
 }
-
+/*
 func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (*Publisher, error) {
 	// read topic configuration
 	topicConfiguration := &messaging_pb.TopicConfiguration{
@@ -24,7 +24,11 @@ func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (*
 	}
 	publishClients := make([]messaging_pb.SeaweedMessaging_PublishClient, topicConfiguration.PartitionCount)
 	for i := 0; i < int(topicConfiguration.PartitionCount); i++ {
-		client, err := mc.setupPublisherClient(namespace, topic, int32(i))
+		client, err := setupPublisherClient(broker.TopicPartition{
+			Namespace: namespace,
+			Topic: topic,
+			Partition: int32(i),
+		})
 		if err != nil {
 			return nil, err
 		}
@@ -35,6 +39,7 @@ func (mc *MessagingClient) NewPublisher(publisherId, namespace, topic string) (*
 		topicConfiguration: topicConfiguration,
 	}, nil
 }
+*/
 
 func setupPublisherClient(grpcConnection *grpc.ClientConn, tp broker.TopicPartition) (messaging_pb.SeaweedMessaging_PublishClient, error) {
 

weed/messaging/msgclient/subscriber.go

@@ -5,6 +5,7 @@ import (
 	"io"
 	"time"
 
+	"google.golang.org/grpc"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
 )
 
@@ -13,6 +14,7 @@ type Subscriber struct {
 	subscriberId      string
 }
 
+/*
 func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string, startTime time.Time) (*Subscriber, error) {
 	// read topic configuration
 	topicConfiguration := &messaging_pb.TopicConfiguration{
@@ -36,9 +38,9 @@ func (mc *MessagingClient) NewSubscriber(subscriberId, namespace, topic string,
 
 func (mc *MessagingClient) setupSubscriberClient(subscriberId, namespace, topic string, partition int32, startTime time.Time) (messaging_pb.SeaweedMessaging_SubscribeClient, error) {
 
-	stream, newBroker, err := mc.initSubscriberClient(subscriberId, namespace, topic, partition, startTime)
+	stream, err := setupSubscriberClient(subscriberId, namespace, topic, partition, startTime)
 	if err != nil {
-		return client, err
+		return stream, err
 	}
 	if newBroker != nil {
 
@@ -47,6 +49,7 @@ func (mc *MessagingClient) setupSubscriberClient(subscriberId, namespace, topic
 	return stream, nil
 
 }
+*/
 
 func setupSubscriberClient(grpcConnection *grpc.ClientConn, subscriberId string, namespace string, topic string, partition int32, startTime time.Time) (stream messaging_pb.SeaweedMessaging_SubscribeClient, err error) {
 	stream, err = messaging_pb.NewSeaweedMessagingClient(grpcConnection).Subscribe(context.Background())
@@ -70,13 +73,10 @@ func setupSubscriberClient(grpcConnection *grpc.ClientConn, subscriberId string,
 	}
 
 	// process init response
-	initResponse, err := stream.Recv()
+	_, err = stream.Recv()
 	if err != nil {
 		return
 	}
-	if initResponse.Redirect != nil {
-		// TODO follow redirection
-	}
 	return stream, nil
 }