async chan write read, no write for closed chan

weed/messaging/broker/broker_grpc_server_publish.go

@@ -1,11 +1,15 @@
 package broker
 
 import (
+	"crypto/md5"
+	"fmt"
 	"io"
 
 	"github.com/golang/protobuf/proto"
 
+	"github.com/chrislusf/seaweedfs/weed/filer2"
 	"github.com/chrislusf/seaweedfs/weed/glog"
+	"github.com/chrislusf/seaweedfs/weed/pb/filer_pb"
 	"github.com/chrislusf/seaweedfs/weed/pb/messaging_pb"
 )
 
@@ -44,9 +48,19 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
 		Topic:     in.Init.Topic,
 		Partition: in.Init.Partition,
 	}
+
+	tpDir := fmt.Sprintf("%s/%s/%s", filer2.TopicsDir, tp.Namespace, tp.Topic)
+	md5File := fmt.Sprintf("p%02d.md5", tp.Partition)
+	// println("chan data stored under", tpDir, "as", md5File)
+
+	if exists, err := filer_pb.Exists(broker, tpDir, md5File, false); err == nil && exists {
+		return fmt.Errorf("channel is already closed")
+	}
+
 	tl := broker.topicLocks.RequestLock(tp, topicConfig, true)
 	defer broker.topicLocks.ReleaseLock(tp, true)
 
+	md5hash := md5.New()
 	// process each message
 	for {
 		// println("recv")
@@ -78,8 +92,16 @@ func (broker *MessageBroker) Publish(stream messaging_pb.SeaweedMessaging_Publis
 			break
 		}
 
+		md5hash.Write(in.Data.Value)
+
 	}
 
+	if err := broker.appendToFile(tpDir+"/"+md5File, topicConfig, md5hash.Sum(nil)); err != nil {
+		glog.V(0).Infof("err writing %s: %v", md5File, err)
+	}
+
+	// fmt.Printf("received md5 %X\n", md5hash.Sum(nil))
+
 	// send the close ack
 	// println("server send ack closing")
 	if err := stream.Send(&messaging_pb.PublishResponse{IsClosed: true}); err != nil {