3 files changed, 44 insertions, 17 deletions
diff --git a/weed/messaging/msgclient/pub_sub_chan.go b/weed/messaging/msgclient/pub_sub_chan.go
index d39e4c658..a11240080 100644
--- a/weed/messaging/msgclient/pub_sub_chan.go
+++ b/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
 		}
 	}()
 
diff --git a/weed/messaging/msgclient/publisher.go b/weed/messaging/msgclient/publisher.go
index b0459494b..08f1d278a 100644
--- a/weed/messaging/msgclient/publisher.go
+++ b/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) {
 
diff --git a/weed/messaging/msgclient/subscriber.go b/weed/messaging/msgclient/subscriber.go
index 27fa35a5b..d3066d6ef 100644
--- a/weed/messaging/msgclient/subscriber.go
+++ b/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
 }