Merge "[FAB-1632] handle panic on closed channel"

core/container/inproccontroller/inprocstream.go

@@ -17,9 +17,18 @@ limitations under the License.
 package inproccontroller
 
 import (
+	"fmt"
+
 	pb "github.com/hyperledger/fabric/protos/peer"
 )
 
+//SendPanicFailure
+type SendPanicFailure string
+
+func (e SendPanicFailure) Error() string {
+	return fmt.Sprintf("send failure %s", string(e))
+}
+
 // PeerChaincodeStream interface for stream between Peer and chaincode instance.
 type inProcStream struct {
 	recv <-chan *pb.ChaincodeMessage
@@ -30,9 +39,17 @@ func newInProcStream(recv <-chan *pb.ChaincodeMessage, send chan<- *pb.Chaincode
 	return &inProcStream{recv, send}
 }
 
-func (s *inProcStream) Send(msg *pb.ChaincodeMessage) error {
+func (s *inProcStream) Send(msg *pb.ChaincodeMessage) (err error) {
+	//send may happen on a closed channel when the system is
+	//shutting down. Just catch the exception and return error
+	defer func() {
+		if r := recover(); r != nil {
+			err = SendPanicFailure(fmt.Sprintf("%s", r))
+			return
+		}
+	}()
 	s.send <- msg
-	return nil
+	return
 }
 
 func (s *inProcStream) Recv() (*pb.ChaincodeMessage, error) {

core/container/inproccontroller/inprocstream_test.go

@@ -0,0 +1,33 @@
+/*
+Copyright IBM Corp. All Rights Reserved.
+
+SPDX-License-Identifier: Apache-2.0
+*/
+
+package inproccontroller
+
+import (
+	"testing"
+
+	pb "github.com/hyperledger/fabric/protos/peer"
+	"github.com/stretchr/testify/assert"
+)
+
+func TestSend(t *testing.T) {
+	ch := make(chan *pb.ChaincodeMessage)
+
+	stream := newInProcStream(ch, ch)
+
+	//good send (non-blocking send and receive)
+	msg := &pb.ChaincodeMessage{}
+	go stream.Send(msg)
+	msg2, _ := stream.Recv()
+	assert.Equal(t, msg, msg2, "send != recv")
+
+	//close the channel
+	close(ch)
+
+	//bad send, should panic, unblock and return error
+	err := stream.Send(msg)
+	assert.NotNil(t, err, "should have errored on panic")
+}